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.

ArchivingStore(*, live: Store, archive: Archive, live_ttl_after_archive: int = 300)

Fields:

• blocking_io: bool
• rebalance_shards: Callable[[str, str], list[int]]
• release_shards: Callable[[str, str], None]

capabilities

capabilities(self) -> frozenset[str]

close

close(self) -> None

Close the archive, then the live store, each only if it has a close.

append_event

append_event(self, name: str, run_id: str, event: RunEvent, *, fence: Optional[str] = None) -> None

append_event_sequenced

append_event_sequenced(self, name: str, run_id: str, event: RunEvent, *, retain: bool = True, fence: Optional[str] = None) -> int

append_events_sequenced

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

append_events

append_events(self, name: str, run_id: str, events: list[RunEvent], *, fence: Optional[str] = None) -> 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]]

read_events_from

read_events_from(self, name: str, run_id: str, from_seq: int, *, limit: Optional[int] = None) -> list[RunEvent]

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

get_chain_tail

get_chain_tail(self, root_name: str, root_run_id: str) -> Optional[ChainLink]

advance_chain_tail

advance_chain_tail(self, link: ChainLink) -> None

close_chain

close_chain(self, link: ChainLink) -> None

seal_inbox

seal_inbox(self, name: str, run_id: str, successor: ChainLink) -> InboxHandoff

import_inbox

import_inbox(self, link: ChainLink, handoff: InboxHandoff) -> None

chain_adopted

chain_adopted(self, name: str, run_id: str) -> bool

read_inbox_seal

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

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) -> tuple[Optional[str], Optional[str]]

lock_holder

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

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

request_pause

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

terminate_run

terminate_run(self, name: str, run_id: str, *, reason: Optional[str] = None, requested_by: Optional[str] = None, cause: Optional[dict[str, Any]] = None, retain: bool = True) -> TerminateOutcome

requeue_run

requeue_run(self, name: str, run_id: str, worker_id: Optional[str] = None, *, reason: str = 'owner_gone') -> 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, branch_options: Optional[dict[str, Any]] = None, headers: Optional[dict[str, str]] = None) -> str

wake_run

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

wake_due

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

time_out_wait

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

reopen_wait

reopen_wait(self, name: str, run_id: str, *, wait_id: str, kind: str, lease: Optional[str] = None) -> WaitOutcome

fail_interrupt

fail_interrupt(self, name: str, run_id: str, *, wait_id: str, code: str, detail: dict[str, Any]) -> WaitOutcome

write_update_result

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

settle_update

settle_update(self, name: str, run_id: str, update_id: str, outcome: dict[str, Any], *, fence: Optional[str] = None, ack: bool = True) -> None

adopt_run

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

get_pause_requested_at

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

get_run_controls

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

get_run_version

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

get_run_rank

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

note_replay_check_timeout

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

hand_back_run

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

register_version

register_version(self, name: str, version: str) -> int

version_rank

version_rank(self, name: str, version: str) -> Optional[int]

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]

---

Batch

A batch of records from a source.

Batch(source_id: str, partition: int, offset: int, records: list[T])

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

---

ChainLink

ChainLink(name: str, run_id: str, root_name: str, root_run_id: str, hop: int)

Fields:

• name: str
• run_id: str
• root_name: str
• root_run_id: str
• hop: int

root

root(cls, name: str, run_id: str) -> Self

successor

successor(self, name: str, run_id: str) -> ChainLink

provenance_source

provenance_source(self) -> dict[str, Any]

continued_payload

continued_payload(self) -> dict[str, Any]

The run:continued payload that hands off to this link.

from_continued_payload

from_continued_payload(cls, payload: Mapping[str, Any]) -> Self

The successor link a run:continued payload names.

Raises ContinuationChainError naming the first key that is missing or the wrong type, so a corrupt journal fails loud instead of surfacing as a KeyError somewhere downstream.

to_json

to_json(self) -> str

from_json

from_json(cls, raw: str | bytes) -> Self

---

ChainStore

The tail record of a continue_as_new chain, keyed by its root.

The record names the newest link the chain has adopted, so any run id in a chain resolves to the live tail in a constant number of reads instead of one read per hop. It only ever moves forward: a late or repeated write never pulls the pointer back to an older link.

get_chain_tail

get_chain_tail(self, root_name: str, root_run_id: str) -> Optional[ChainLink]

The newest adopted link of the chain rooted at this run, or None.

advance_chain_tail

advance_chain_tail(self, link: ChainLink) -> None

Record link as the chain’s tail unless a later hop is already recorded.

close_chain

close_chain(self, link: ChainLink) -> None

Record link as the chain’s final link (unless a later hop is recorded) and let the record expire with the run’s other retained keys.

The built-in stores also do this when the final link’s terminal event is appended, whoever writes it, so a call after that only finishes any cleanup still pending.

seal_inbox

seal_inbox(self, name: str, run_id: str, successor: ChainLink) -> InboxHandoff

Freeze run_id’s inbox for successor and return what it held.

Sends addressed to the sealed run are redirected to successor from here on. Sealing again for the same successor returns the same snapshot; sealing for a different one raises ContinuationChainError.

import_inbox

import_inbox(self, link: ChainLink, handoff: InboxHandoff) -> None

Merge handoff into link’s inbox, ahead of what it already holds, and mark link adopted. Idempotent.

chain_adopted

chain_adopted(self, name: str, run_id: str) -> bool

Whether this successor has already imported its predecessor’s inbox.

read_inbox_seal

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

The successor this run’s inbox is sealed to, or 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.

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

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.

CompositeStore(*, lifecycle: LifecycleStore, events: EventLogStore)

Fields:

• blocking_io: bool
• read_events_from: Callable[..., list[RunEvent]] The events backend’s method, or AttributeError when it keeps no run_seq: the engine detects suffix reads with getattr, and a method defined on the class would claim them over any backend.

capabilities

capabilities(self) -> frozenset[str]

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, branch_options: Optional[dict[str, Any]] = None, headers: Optional[dict[str, str]] = 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, *, reason: str = 'owner_gone') -> 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) -> tuple[Optional[str], Optional[str]]

request_cancel

request_cancel(self, name: str, run_id: str, *, detail: Optional[dict[str, Any]] = None) -> 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]

get_run_controls

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

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

lock_holder

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

append_event

append_event(self, name: str, run_id: str, event: RunEvent, *, fence: Optional[str] = None) -> None

append_events

append_events(self, name: str, run_id: str, events: list[RunEvent], *, fence: Optional[str] = None) -> 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

fail_interrupt

fail_interrupt(self, name: str, run_id: str, *, wait_id: str, code: str, detail: dict[str, Any]) -> WaitOutcome

hand_back_run

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

terminate_run

terminate_run(self, name: str, run_id: str, *, reason: Optional[str] = None, requested_by: Optional[str] = None, cause: Optional[dict[str, Any]] = None, retain: bool = True) -> TerminateOutcome

get_run_version

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

adopt_run

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

note_replay_check_timeout

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

register_version

register_version(self, name: str, version: str) -> int

version_rank

version_rank(self, name: str, version: str) -> Optional[int]

get_run_rank

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

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

wake_due

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

time_out_wait

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

reopen_wait

reopen_wait(self, name: str, run_id: str, *, wait_id: str, kind: str, lease: Optional[str] = None) -> 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, wait_id: Optional[str] = None) -> 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

settle_update

settle_update(self, name: str, run_id: str, update_id: str, outcome: dict[str, Any], *, fence: Optional[str] = None, ack: bool = True) -> None

read_update_result

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

get_chain_tail

get_chain_tail(self, root_name: str, root_run_id: str) -> Optional[ChainLink]

advance_chain_tail

advance_chain_tail(self, link: ChainLink) -> None

close_chain

close_chain(self, link: ChainLink) -> None

seal_inbox

seal_inbox(self, name: str, run_id: str, successor: ChainLink) -> InboxHandoff

import_inbox

import_inbox(self, link: ChainLink, handoff: InboxHandoff) -> None

chain_adopted

chain_adopted(self, name: str, run_id: str) -> bool

read_inbox_seal

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

---

ContinuationChainError

Raised when a continuation chain cannot be followed or joined. Either a run:continued link is unreadable (a missing field, or one of the wrong type), or the chain’s own bookkeeping refuses: an inbox already sealed to a different successor, a run continued with no predecessor, or inbox redirects that did not settle. reason says which. Fails loud rather than surface a raw KeyError/TypeError from a corrupt journal.

ContinuationChainError(name: str, run_id: str, *, reason: str)

Fields:

• name: str
• run_id: str
• reason: str

---

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.

Debounce(key: str | Callable[[dict], str], window_seconds: float)

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

DurableInterrupt(name: str, run_id: str, event_id: Optional[str], payload: dict[str, Any], *, wait_id: Optional[str] = None)

Fields:

• code = ErrorCode.SUSPEND
• name: str
• run_id: str
• event_id: Optional[str]
• payload: dict[str, Any]
• wait_id: Optional[str]

---

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.

DurableInvocationCancelled(name: str, run_id: str, detail: str)

Fields:

• code = ErrorCode.CANCELLED

---

DurableInvocationError

A run driven by run_and_await ended in error.

DurableInvocationError(name: str, run_id: str, detail: str)

Fields:

• name: str
• run_id: str
• detail: str

---

DurableInvocationTimeout

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

DurableInvocationTimeout(name: str, run_id: str, detail: str)

Fields:

• code = ErrorCode.RUN_TIMEOUT

---

EventKind

Core event kinds: lifecycle, step and the per-call model cost.

Agent-specific kinds (LLM_START, BLOCK_, TOOL_, AGENT_CALL_*) live in pyrula.agents’s EventKind and ride the wire as plain strings. llm:done is core because a workflow run writes it too, for a journaled model call made through a drop-in client, and every reader of a workflow journal has to accept it.

The set is open: a release may add kinds, and RunEvent.kind is a plain str that can hold a kind this enum does not list (including your own from ctx.emit). Compare by value and never match exhaustively.

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_WAIT_TIMED_OUT = 'run:wait_timed_out'
• 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'
• LLM_DONE = 'llm:done'
• 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, *, fence: Optional[str] = None) -> None

Append event to the run’s journal.

fence is the lease of the run’s holder. A fenced append commits only while that lease still holds the run’s lock and the journal has not ended (its last event is not terminal); otherwise it raises LockLost and commits nothing. The run engine fences every append it makes as the holder; writers that do not hold the run (signals, wakes, requeue) leave fence as None.

append_events

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

Append events in order, fenced as append_event. Default loops over append_event.

A batch with an event behind a terminal one is refused whole, with EventAfterTerminal (a LockLost subclass; a plain LockLost means a lost lease), and nothing of it is written (refuse_event_after_terminal).

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.

code is the caller’s own identifier for the failure. It ends up in run:error.user_code and is shipped at every capture level, so keep it an identifier, never data.

Fatal(*args: object, code: Optional[str] = None)

Fields:

• user_code: Optional[str]

---

IdempotencyConflict

An idempotency key was reused with a different request: another start_at or delay. The key’s first run is unchanged.

Fields:

• code = ErrorCode.IDEMPOTENCY_CONFLICT

---

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.

InMemorySource(*, source_id: str = 'test-topic', records: Optional[list[T]] = None, batch_size: int = 10)

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.

---

InboxHandoff

What a continued run’s successor takes over: the sealed inbox.

Signals are per-name FIFO, each entry carrying its name and the idempotency key it was sent with. signal_dedupe maps an idempotency key to the store’s own record of the send, which only the store that wrote it reads back.

InboxHandoff(predecessor_name: str, predecessor_run_id: str, signals: tuple[dict[str, Any], ...] = (), updates: tuple[dict[str, Any], ...] = (), update_ids: tuple[str, ...] = (), signal_dedupe: Mapping[str, str] = <factory>, update_results: tuple[CarriedUpdateResult, ...] = (), cancel_requested_at: Optional[str] = None, pause_requested_at: Optional[str] = None)

Fields:

• predecessor_name: str
• predecessor_run_id: str
• signals: tuple[dict[str, Any], ...] = ()
• updates: tuple[dict[str, Any], ...] = ()
• update_ids: tuple[str, ...] = ()
• signal_dedupe: Mapping[str, str] = <factory>
• update_results: tuple[CarriedUpdateResult, ...] = ()
• cancel_requested_at: Optional[str] = None
• pause_requested_at: Optional[str] = None

---

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.

gate_timeout bounds how long run() waits for a concurrency or rate limit to make room before it raises RunGatedError.

Call start() when the host starts, so the runner registers with Cloud and can receive controls and cases before its first run; run() and run_next() otherwise connect on first use. Pair it with aclose().

InlineWorkflowRunner(*, store: Store, worker_id: Optional[str] = None, cloud: bool = True, live_events: Optional[bool] = None, clock: Callable[[], float] = time.time, reject_continue_as_new: bool = False, capture: Optional[CaptureLevel | str] = None, case_sessions: Optional[CaseSessionFactory] = None, gate_timeout: Optional[float] = GATE_TIMEOUT_S)

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.

start

start(self) -> None

Connect to Cloud and register this runner’s workflows.

A no-op without cloud=True and PYRULA_CLOUD_API_KEY, and on a second call. Raises ConfigurationError when a key is set but the cloud client is not installed; any other connect failure is logged and the runner continues without cloud.

run_next

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

run

run(self, name: str, run_id: str, *, lease: Optional[str] = None, _strict_store: bool = False) -> Optional[list[RunEvent]]

Advance run_id here by one attempt, under lease (a fresh one if not given), and return the run’s public events.

When another live claim holds the run, this waits for that holder and returns the run’s events once it ends. If the holder’s lock lapses first (it died), the run is claimed here under a fresh lease and resumed.

The run passes its concurrency and rate limits as a worker’s claim does: while its Limit has no free slot or its Rate no token, this waits for room rather than running it over the limit, and raises RunGatedError after gate_timeout seconds (300 by default; None waits without limit). A run submitted earlier under the same limit holds its room until it is run or cancelled, so every gated run submitted to an inline runner has to be driven. A wait longer than ten seconds is logged once, naming the gate.

Raises StoreUnavailable when the store failed under the attempt. The run is left non-terminal, with no run:error, so calling run again resumes it from its journal.

This differs from pyrula.agents’ InlineWorker.run_turn, which raises RunOwnedError instead of waiting - an intentional asymmetry between the two embedded runners, not a bug in either one.

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

InterruptResponse(action: InterruptAction | str, value: Any = None)

Fields:

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

---

JournalNeedsNewerRuntime

A journal holds facts this runtime cannot read, so it must not replay it.

missing names each unreadable thing (an event kind, a step kind, a required feature, an encoding, a schema major) so an operator can tell what to upgrade for. The run is not broken: a worker that knows the facts can replay it.

JournalNeedsNewerRuntime(message: Optional[str] = None, *, missing: Sequence[str] = ())

Fields:

• missing

---

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.

blocking_io says whether a call can block on I/O. It defaults to True, so the async runtime reads run controls on a worker thread. A store whose calls never block (in-process state only) sets it to False and is called inline, with no thread hop per boundary. A wrapper reports its inner store’s value.

Fields:

• blocking_io: bool = True

capabilities

capabilities(self) -> frozenset[str]

The optional groups this store provides, by name (see CAPABILITY_GROUPS).

A group counts when the store’s class overrides every method of it, so a method that is still an ABC default does not. A store that forwards to another overrides this and reports what it can serve.

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, branch_options: Optional[dict[str, Any]] = None, headers: Optional[dict[str, str]] = 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.

branch_options is a counterfactual branch’s own overrides, written into its run:init under branch_options (local only, never forwarded to a collector) so a worker that resumes the branch restores them. Absent for every other run. A store whose submit_run lacks the keyword is not passed it (the callers and the forwarding stores check the bound method with accepts_branch_options and warn once per store class); a branch started on it resumes without its overrides, as it did before.

headers are stamped on the run:init event, atomically with the run: the requires header a journal fact that changes what replay means must carry on its first event. They are forwarded only when set, and a store whose submit_run lacks the keyword raises TypeError rather than drop them, because a worker that never saw the header would replay the run wrong.

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.

A queued run whose lock another lease holds (woken or signalled while its holder finishes the attempt) is not eligible: it is passed over, keeps its place in the queue, and does not stop the runs behind it from being claimed. A store may bound how many such runs one claim looks past (the Valkey store looks at 64 queued runs per version bucket and shard); when every run inside the bound is held, the claim takes nothing until one lets go.

claim_run

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

Take the named run for the lease worker_id (lock + deadline).

Distinct from claim_next_run (consume the queue): this claims the named run_id directly. It succeeds when the run’s lock is free or already held by the same lease, and raises RunOwnedError when another lease holds it. A run that is still pending leaves the queue, so claim_next_run cannot hand it out again.

Taken from a free lock, the run passes its concurrency and rate gates as a queue claim does: it takes a slot and spends a token, or raises RunGatedError and stays as it was. Re-asserting a claim the lease already holds takes neither.

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, *, reason: str = 'owner_gone') -> None

Put a claimed run back on the queue and record a run:requeued event carrying reason. owner_gone (a crashed or vanished attempt) counts against the store’s max_retries; store_unavailable (the attempt lost the store) counts against max_store_requeues. Each budget counts the requeues since the run last made progress (a step:done or a handled update), so outages spread over a long life do not add up; an attempt that crashes on the same step every time does. Once the matching budget is spent the call ends the run instead (max_retries or store_unavailable) and releases it. With worker_id set, does nothing unless that lease still holds the run. A run whose journal already ends in run:complete or run:continued, or for an owned requeue in run:error too (the reply to the attempt’s terminal append was lost), is only released: queued again, it would run and end a second time. An unowned requeue of a failed run retries it.

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) -> tuple[Optional[str], Optional[str]]

lock_holder

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

The lease currently holding the run’s lock, or None when no lock is held (never claimed, released, or expired for want of a heartbeat).

request_cancel

request_cancel(self, name: str, run_id: str, *, detail: Optional[dict[str, Any]] = None) -> str

Latch a graceful cancel and audit it once with a run:cancel event.

detail rides on that event and becomes the detail of the cancelled terminal the run’s worker writes (a parent cancelling a child passes requested_by and cause). Only the first request records it: a cancel already latched keeps its own.

get_cancel_requested_at

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

get_run_controls

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

(cancel_requested_at, pause_requested_at) in one read: what a running attempt polls before each step, so it must not cost two round trips.

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

Re-stamp an in-flight run onto new_version (compatible upgrade). The run keeps its version_behavior, and an auto_upgrade run takes the new version’s rank, so it can move forward again and never backward.

With worker_id set, re-stamps only while that lease still holds the run, so a worker whose lease lapsed during its replay check cannot adopt. Returns False when the lease no longer holds it and nothing changed, else True.

note_replay_check_timeout

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

Count one replay check of this run that timed out and return how many it has had since it was last adopted; 0 when worker_id is given and that lease no longer holds the run. Persisted with the run, so the count survives a worker restart, and cleared by adopt_run. The caller ends the run once it reaches MAX_REPLAY_CHECK_DEFERRALS.

The default is for a backend with nowhere to keep the count: it reports the bound as reached, so the first timeout ends the run instead of handing it back forever, and logs that once.

register_version

register_version(self, name: str, version: str) -> int

Return the rank of version for workflow name, assigning the next one on first sight. A rank is a per-name integer fixed for the life of the store: a higher rank is a version seen later, which is how a rolling deploy orders code. Idempotent, and safe to call from many workers at once.

version_rank

version_rank(self, name: str, version: str) -> Optional[int]

The rank register_version assigned, or None for a version never seen.

get_run_rank

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

The rank an auto_upgrade run was last stamped with (its submit, or the adoption that moved it), or None for a run with no rank: a pinned run, or one stored before ranks existed. A claim treats None as the oldest rank.

hand_back_run

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

Give a claim back unrun: drop the lock and concurrency slot and put the run back in its own bucket, with no run:requeued event and no retry count.

For a worker that declined the run (newer code owns it), as opposed to requeue_run, which records a crashed attempt and counts toward max_retries. With worker_id set, does nothing unless that lease still holds the run.

Optional (capability hand_back). Without it the worker releases its claim and leaves the run to lease recovery, which does count an attempt.

terminate_run

terminate_run(self, name: str, run_id: str, *, reason: Optional[str] = None, requested_by: Optional[str] = None, cause: Optional[dict[str, Any]] = None, retain: bool = True) -> TerminateOutcome

End the run now, without its worker.

In one atomic step: append run:error with code: "terminated" and detail {terminated_at, reason, requested_by, cause}, latch the cancel flag, delete the run’s lock, and take the run out of the queue, the sleeping, signal-waiting and interrupted sets, giving back its concurrency slot and admission. The deleted lock is the fence: a holder that is still running, or comes back, has every leased append refused with LockLost and its next heartbeat refused. A run sealed to a successor is followed to the chain’s live tail, as request_cancel does.

Returns ALREADY_ENDED (writing nothing) when the journal already ends, and NOT_FOUND for an id with no run. cause names the parent that forced a child to end. retain=False leaves the completion TTL off, as for an append, for a store that archives the run before it expires.

Optional (capability terminate). A store without it cannot terminate a run.

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. The default reads it from get_run_controls.

request_pause

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

Record a pause request for the run and return its ISO timestamp. Idempotent while the request is open. Pairs with request_cancel.

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.

fail_interrupt

fail_interrupt(self, name: str, run_id: str, *, wait_id: str, code: str, detail: dict[str, Any]) -> WaitOutcome

End the run as failed with code because its on_interrupt handler could not deliver the question, only if the interrupt named wait_id is still open: latches a cancel carrying code and detail and queues the run so a worker writes the error. A closed interrupt answers SUPERSEDED.

Optional (capability interrupt_fail). Without it the engine logs the failure and leaves the run parked on the question.

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

Limit(key: str | Callable[[dict], str], max: int)

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.

MemoryStore(*, heartbeat_interval: float = 10.0, lock_ttl: float = 30.0, held_ttl: float = 86400.0, max_replay_entries: int = 200000, max_retries: int = 3, max_store_requeues: int = 20)

Fields:

• blocking_io: bool = False
• heartbeat_interval: float
• lock_ttl: float
• held_ttl: float
• max_replay_entries: int
• max_retries: int
• max_store_requeues: int

list_names

list_names(self) -> list[str]

---

NestedDurableOperationError

Durable orchestration attempted inside an atomic step or tool.

NestedDurableOperationError(operation: str, operation_name: str, boundary_kind: str, boundary_name: str)

Fields:

• operation: str
• operation_name: str
• boundary_kind: str
• boundary_name: 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.

code is the caller’s own identifier for the failure. It ends up in run:error.user_code and is shipped at every capture level, so keep it an identifier, never data.

Poison(*args: object, code: Optional[str] = None)

Fields:

• user_code: Optional[str]

---

PostgresArchive

Postgres-backed durable archive for completed runs.

Needs the pyrula-workflows[postgres] extra (asyncpg, alembic, sqlalchemy).

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 reads the schema’s compat row and raises ArchiveSchemaMismatch (not StoreUnavailable) unless this pyrula version can run on it - a missing, foreign or too-new schema is a deploy error, not a condition worth retrying. check_schema() runs the same check on demand.

PostgresArchive(*, dsn: str, replay_page_size: int = 1000, pool: Optional[Any] = None, max_connections: int = 8)

Fields:

• dsn: str
• replay_page_size: int

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 its compat row is not readable by this pyrula version; returns None when it is.

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.

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

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

ReplayState(recorded_call_results: dict[tuple[str, int], Any] = <factory>, recorded_call_results_by_id: dict[tuple[str, str], Any] = <factory>, recorded_call_fingerprints: dict[tuple[str, int], str] = <factory>, recorded_call_fingerprints_by_id: dict[tuple[str, str], str] = <factory>, recorded_call_slots: dict[str, list[int]] = <factory>, recorded_call_failures: dict[tuple[str, int], tuple[str, int]] = <factory>, recorded_call_reservations: dict[tuple[str, int], str] = <factory>, recorded_call_questions: dict[tuple[str, int], tuple[str, int]] = <factory>, manual_checkpoint_counts: dict[str, int] = <factory>, emit_counts: dict[str, int] = <factory>, fanout_results: dict[str, list[Any]] = <factory>, fanout_fingerprints: dict[str, str] = <factory>, fanout_pending: dict[str, list[dict[str, str]]] = <factory>, fanout_pending_fingerprints: dict[str, str] = <factory>, child_results: dict[tuple[str, int], Any] = <factory>, signal_send_results: dict[str, Any] = <factory>, signal_send_fingerprints: dict[str, str] = <factory>, deterministic_results: dict[tuple[str, int], Any] = <factory>, interrupt_results: dict[tuple[str, int], dict[str, Any]] = <factory>, interrupt_fingerprints: dict[tuple[str, int], str] = <factory>, sleep_results: dict[str, set[int]] = <factory>, sleep_armed: dict[tuple[str, int], str] = <factory>, sleep_wait_ids: dict[tuple[str, int], str] = <factory>, open_sleep: Optional[tuple[str, int]] = None, signal_armed: dict[tuple[str, int], tuple[str, Optional[str]]] = <factory>, interrupt_armed: dict[tuple[str, int], tuple[str, Optional[str]]] = <factory>, signal_results: dict[tuple[str, int], dict[str, Any]] = <factory>, signal_fingerprints: dict[tuple[str, int], str] = <factory>, journaled_signal_ids: set[str] = <factory>, named_step_results: dict[tuple[str, int], Any] = <factory>, named_step_errors: dict[tuple[str, int], list[dict[str, Any]]] = <factory>, last_step_error: Optional[tuple[tuple[str, int], int]] = None, escaped_step_error: Optional[tuple[tuple[str, int], int]] = None, named_step_fingerprints: dict[tuple[str, int], str] = <factory>, child_fingerprints: dict[tuple[str, int], str] = <factory>, captured_step_keys: set[tuple[str, int]] = <factory>, captured_call_keys: set[tuple[str, str]] = <factory>, stream_committed_offsets: dict[tuple[str, int], int] = <factory>, stream_committed_batch_counts: dict[tuple[str, int], int] = <factory>, stream_carry: dict[str, Any] = <factory>, patch_markers: set[str] = <factory>, updates_by_frontier: dict[int, list[dict[str, Any]]] = <factory>, journaled_update_ids: set[str] = <factory>, journaled_update_outcomes: dict[str, dict[str, Any]] = <factory>, journaled_effect_count: int = 0)

Fields:

• recorded_call_results: dict[tuple[str, int], Any] = <factory>
• recorded_call_results_by_id: dict[tuple[str, str], Any] = <factory>
• recorded_call_fingerprints: dict[tuple[str, int], str] = <factory>
• recorded_call_fingerprints_by_id: dict[tuple[str, str], str] = <factory>
• recorded_call_slots: dict[str, list[int]] = <factory>
• recorded_call_failures: dict[tuple[str, int], tuple[str, int]] = <factory>
• recorded_call_reservations: dict[tuple[str, int], str] = <factory>
• recorded_call_questions: dict[tuple[str, int], tuple[str, int]] = <factory>
• manual_checkpoint_counts: dict[str, int] = <factory>
• emit_counts: dict[str, int] = <factory>
• fanout_results: dict[str, list[Any]] = <factory>
• fanout_fingerprints: dict[str, str] = <factory>
• fanout_pending: dict[str, list[dict[str, str]]] = <factory>
• fanout_pending_fingerprints: dict[str, str] = <factory>
• child_results: dict[tuple[str, int], Any] = <factory>
• signal_send_results: dict[str, Any] = <factory>
• signal_send_fingerprints: dict[str, str] = <factory>
• deterministic_results: dict[tuple[str, int], Any] = <factory>
• interrupt_results: dict[tuple[str, int], dict[str, Any]] = <factory>
• interrupt_fingerprints: dict[tuple[str, int], str] = <factory>
• sleep_results: dict[str, set[int]] = <factory>
• sleep_armed: dict[tuple[str, int], str] = <factory>
• sleep_wait_ids: dict[tuple[str, int], str] = <factory>
• open_sleep: Optional[tuple[str, int]] = None
• signal_armed: dict[tuple[str, int], tuple[str, Optional[str]]] = <factory>
• interrupt_armed: dict[tuple[str, int], tuple[str, Optional[str]]] = <factory>
• signal_results: dict[tuple[str, int], dict[str, Any]] = <factory>
• signal_fingerprints: dict[tuple[str, int], str] = <factory>
• journaled_signal_ids: set[str] = <factory>
• named_step_results: dict[tuple[str, int], Any] = <factory>
• named_step_errors: dict[tuple[str, int], list[dict[str, Any]]] = <factory>
• last_step_error: Optional[tuple[tuple[str, int], int]] = None
• escaped_step_error: Optional[tuple[tuple[str, int], int]] = None
• named_step_fingerprints: dict[tuple[str, int], str] = <factory>
• child_fingerprints: dict[tuple[str, int], str] = <factory>
• captured_step_keys: set[tuple[str, int]] = <factory>
• captured_call_keys: set[tuple[str, str]] = <factory>
• stream_committed_offsets: dict[tuple[str, int], int] = <factory>
• stream_committed_batch_counts: dict[tuple[str, int], int] = <factory>
• stream_carry: dict[str, Any] = <factory>
• patch_markers: set[str] = <factory>
• updates_by_frontier: dict[int, list[dict[str, Any]]] = <factory>
• journaled_update_ids: set[str] = <factory>
• journaled_update_outcomes: dict[str, dict[str, Any]] = <factory>
• journaled_effect_count: int = 0

---

RetryPlan

What retrying one failed run would copy and start.

cut is the index in the source journal where the retry’s own work begins: the step failure that escaped the body, else the terminal run:error. seed is the source’s events before it, without the source’s run:init. in_doubt names write steps that were started and had no recorded outcome; the retry runs each once.

RetryPlan(name: str, source_run_id: str, retry_run_id: str, cut: int, seed: tuple[RunEvent, ...], seeded_events: int, in_doubt: tuple[str, ...], attempt: int, params: dict[str, Any] = <factory>, metadata: dict[str, Any] = <factory>, version: str = '')

Fields:

• name: str
• source_run_id: str
• retry_run_id: str
• cut: int
• seed: tuple[RunEvent, ...]
• seeded_events: int
• in_doubt: tuple[str, ...]
• attempt: int
• params: dict[str, Any] = <factory>
• metadata: dict[str, Any] = <factory>
• version: str = ''

provenance

provenance(self) -> dict[str, Any]

---

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. jitter is the fraction of each delay that is randomized: the wait is d * (1 - jitter) plus a uniform draw up to d * jitter (0 = exact delays, 1 = full jitter), so workers retrying one throttled provider do not retry in step. 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.

RetryPolicy(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, jitter: float = 0.5)

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
• jitter: float = 0.5

should_retry

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

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

Poison/Fatal, and any error that sets never_retried (a spend cap), 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, rng: Optional[random.Random] = None) -> float

Delay before the attempt after attempt (1-based), jittered after the cap.

rng is for tests; the default is a private generator that user seeding of the global random module cannot affect.

---

RetryRefused

A run that cannot be retried, with the stable code saying why. Raised before anything is written.

RetryRefused(code: ErrorCode, message: Optional[str] = None)

Fields:

• code: ErrorCode

---

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.

headers is the envelope’s extension point: string facts about the event that a later release can add without changing the field set, which is closed at kind, payload, run_id, schema_version, run_seq and headers. Every store round-trips it; an absent value reads as empty. Only the framework sets it.

RunEvent(kind: str, payload: dict[str, Any], run_id: str, schema_version: str = EVENT_SCHEMA_VERSION, run_seq: Optional[int] = None, headers: Mapping[str, str] = <factory>)

Fields:

• kind: str
• payload: dict[str, Any]
• run_id: str
• schema_version: str = EVENT_SCHEMA_VERSION
• run_seq: Optional[int] = None
• headers: Mapping[str, str] = <factory>

---

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.

RunHandle(client: WorkflowClient, name: str, run_id: str)

Fields:

• name: str
• run_id: str

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, *, reason: Optional[str] = None, requested_by: Optional[str] = None) -> None

terminate

terminate(self, *, reason: Optional[str] = None, requested_by: Optional[str] = None) -> TerminateResult

retry

retry(self) -> RunHandle

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

---

RunNotStartedError

An update named a run whose start is still ahead (submit(start_at=, delay=)). Its handlers are registered by the body, so there is nothing to apply it to yet.

Fields:

• code = ErrorCode.INVALID_REQUEST

---

RunNotWaitingError

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

Fields:

• code = ErrorCode.INVALID_REQUEST

---

RunRef

RunRef(name: str, run_id: str)

Fields:

• name: str
• run_id: str

---

RunStatus

Where a run is in its life. The set is open: a release may add statuses, so compare by value and never match exhaustively.

Fields:

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

---

SQLiteStore

Durable store in a single SQLite file.

The constructor opens the file and refuses a path that names no file, a network filesystem, a second host, Windows, an old SQLite library and a file written by a newer pyrula.

retention_seconds is how long a finished run is kept (None keeps it for good); run_maintenance, which the worker calls, deletes it after that. held_ttl is how long a cancel or pause for an id not yet submitted waits for its submit before it is dropped. clock is the time source for leases, retention and the host guard; every process on one file must use the same one, so anything but the default is for tests.

SQLiteStore(path: str | os.PathLike[str], *, retention_seconds: Optional[int] = DEFAULT_RETENTION_SECONDS, synchronous: Literal['full', 'normal'] = 'full', busy_timeout_s: Optional[float] = None, lock_ttl: float = 30.0, held_ttl: float = 86400.0, heartbeat_interval: Optional[float] = None, max_replay_entries: int = 200000, max_retries: int = 3, max_store_requeues: int = 20, clock: Callable[[], float] = time.time)

Fields:

• blocking_io: bool = True
• load_run = StoreHelpers.load_run
• kv_search = StoreHelpers.kv_search
• expire_run = StoreHelpers.expire_run
• get_interrupt_response = StoreHelpers.get_interrupt_response
• path: str | os.PathLike[str]
• retention_seconds: Optional[int]
• busy_timeout_s: Optional[float]
• lock_ttl: float
• held_ttl: float
• heartbeat_interval: Optional[float]
• max_replay_entries: int
• max_retries: int
• max_store_requeues: int

from_url

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

Open the file a sqlite:///relative.db or sqlite:////absolute.db URL names; its synchronous and retention_seconds parameters are options.

backup

backup(self, dest: str | os.PathLike[str]) -> None

Write a consistent, compacted copy of the store to dest while workers keep running. dest must be a new file; it is created owner-only. Copying the .db file alone is not a backup: committed writes can still be in -wal.

info

info(self) -> dict[str, Any]

The file’s path, size and write-ahead log size, its format generation, the retention, run counts by status and the hosts that have opened it.

ping

ping(self) -> None

list_names

list_names(self) -> list[str]

close

close(self) -> None

---

ScheduleBackend

schedule_create

schedule_create(self, schedule: Schedule) -> None

schedule_get

schedule_get(self, schedule_id: str) -> Optional[Schedule]

schedule_list

schedule_list(self, name: Optional[str] = None) -> list[Schedule]

schedule_update

schedule_update(self, schedule: Schedule) -> None

schedule_advance

schedule_advance(self, fired: Schedule, updates: dict[str, Any]) -> bool

schedule_patch

schedule_patch(self, schedule_id: str, mutate: Callable[[Schedule], Optional[Schedule]]) -> Optional[Schedule]

Read, mutate and write one record atomically: no write between the read and the write is lost. mutate gets a copy and returns the new record, or None for no change; an exception it raises writes nothing and propagates. It may run more than once, so it must not have side effects. Returns the stored record, or None when it is missing.

schedule_delete

schedule_delete(self, schedule_id: str) -> bool

schedule_due

schedule_due(self, now: float) -> list[Schedule]

schedule_claim_fire

schedule_claim_fire(self, schedule_id: str, slot: int) -> bool

---

SchemaVersionMismatch

An event’s schema_version major is newer than the runtime understands.

SchemaVersionMismatch(message: Optional[str] = None, *, missing: Sequence[str] = ())

---

Signal

Signal(name: str, payload: dict[str, Any], sent_at: str)

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

With wait_id (a re-claimed attempt re-arming a journaled wait) the peek first checks that wait is still open. If a timeout or a delivery already closed it, the result is {"wait_closed": True} and nothing is armed, so one wait never gets two outcomes.

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.

StepContext(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, _is_cancelled: Callable[[], bool])

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

What a journaled step was. The set is open: a release may add kinds, so compare by value and never match exhaustively.

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.

StepResultTypeError(step_id: str, validation_error: Exception)

Fields:

• step_id: str
• validation_error: Exception

---

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.

StoreBinding(project_id: str, environment: str, bound_by: str, bound_at: str)

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

StoreItem(key: str, value: dict[str, Any], updated_at: float)

Fields:

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

---

TerminateResult

What WorkflowClient.terminate did.

outcome is TERMINATED when this call ended the run and ALREADY_ENDED when the run’s journal already ended (a repeat, or a run that finished first). children counts the open child runs the cascade closed after escalation: asked to cancel, terminated, and left running by their abandon policy; a child that had already ended, or was already asked to cancel, is not counted. run_id is the run that was ended, which for a continue_as_new chain is the live tail rather than the id the caller named. ended holds (name, run_id) of each child and grandchild the cascade terminated: the store wrote their terminal, so no worker reports it.

TerminateResult(outcome: TerminateOutcome, children: CloseResult, run_id: str, ended: tuple[tuple[str, str], ...] = ())

Fields:

• outcome: TerminateOutcome
• children: CloseResult
• run_id: str
• ended: tuple[tuple[str, str], ...] = ()

---

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, lease: 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.

lease is the claim of the attempt running the run, closing a sleep it has reached after the wake time. The call is fenced like an append, raising LockLost unless that lease holds the run’s lock and the run has not ended, and it does not queue the run: it is already executing.

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.

A signal wait or interrupt with a journaled deadline is ended the same way, with run:wait_timed_out. A signal wait whose signal is already buffered, and a wait with no deadline, are left to their own paths and answer SUPERSEDED.

time_out_wait

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

End the signal wait or interrupt named wait_id at its deadline, only if it is still open and, for a signal wait, no signal for it is buffered. lease is the attempt that reached the deadline while it held the run, fenced like wake_run and not queueing the run.

Optional (capability wait_timeout). Without it a holder does not close its own expired wait, and the timer’s wake_due ends it instead.

reopen_wait

reopen_wait(self, name: str, run_id: str, *, wait_id: str, kind: str, lease: Optional[str] = None) -> WaitOutcome

Journal the opening of an unanswered signal wait (kind="signal") or interrupt ("interrupt") once more, copied from the original with reopened set, for an attempt that parks on it again, so the run’s last event says it is waiting. Written only while wait_id is open, in the same step that checks it: RESOLVED when written, DUPLICATE when the last event already is this opening, SUPERSEDED when the wait is closed (the caller writes nothing more and the queued re-claim replays the outcome). lease fences it like an append.

Optional (capability wait_reopen). Without it an attempt parks on the wait without journaling its opening again.

---

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.

cluster=None (the default) asks the server through INFO whether it is a cluster and builds the client that fits; pass True or False to override, which a cluster behind a proxy that hides it needs. An ElastiCache Serverless host is always a cluster and is not asked. A server that does not answer INFO on any attempt is down, and construction raises StoreUnavailable instead of trying the same address again to load the script.

ValkeyStore(*, url: str, completed_ttl: int = 86400, maxlen: int = 100000, replay_page_size: int = 1000, replay_page_bytes: int = 8 * 1024 * 1024, heartbeat_interval: float = 10.0, lock_ttl: float = 30.0, max_replay_entries: Optional[int] = None, retry_attempts: int = 3, retry_base: float = 0.1, max_retries: int = 3, max_store_requeues: int = 20, socket_timeout: float = 5.0, socket_connect_timeout: float = 2.0, shard_count: int = SHARD_COUNT_DEFAULT, shard_lease_ttl: float = 30.0, cluster: Optional[bool] = None, namespace: Optional[str] = None, eviction_policy_check: Optional[str] = None)

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: str
• completed_ttl: int
• maxlen: int
• max_run_event_count: Optional[int]
• replay_page_size: int
• replay_page_bytes: int
• heartbeat_interval: float
• lock_ttl: float
• max_replay_entries
• retry_attempts: int
• retry_base: float
• max_retries: int
• max_store_requeues: int
• socket_timeout: float
• socket_connect_timeout: float
• shard_count: int
• shard_lease_ttl: float
• cluster: bool

from_url

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

resolve_cluster

resolve_cluster(url: str, probe: Callable[[], Any]) -> bool

Whether url names a cluster. ElastiCache Serverless always does and is not asked. Anything else is what probe(), an INFO answer, reports: redis_mode says cluster, or cluster_enabled is 1 (newer Valkey has no redis_mode, only server_mode). A probe that fails or answers something else means a plain server, as every store assumed before detection; a cluster that slips through is named by the first MOVED or ASK it answers.

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.

---

WaitTimeout

A wait_for_signal or interrupt reached its deadline with no answer.

An ordinary exception, unlike the suspension signals: the body is running again when it is raised and may catch it and carry on. Uncaught, the run ends with wait_timeout.

WaitTimeout(wait_kind: str, name: str, call_index: Optional[int] = None, deadline: Optional[datetime] = None, message: Optional[str] = None)

Fields:

• code = ErrorCode.WAIT_TIMEOUT
• wait_kind: str
• name: str
• call_index: Optional[int]
• deadline: Optional[datetime]

---

WorkerOptions

Tuning knobs for WorkflowWorker

WorkerOptions(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, poison_counts_loop_blocked: bool = True, 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, event_buffer_size: int = 64, step_boundary_durability: bool = True, journal_cache_events: int = 50000, health_port: Optional[int] = None, health_host: str = '127.0.0.1')

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
• poison_counts_loop_blocked: bool = True
• 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
• event_buffer_size: int = 64
• step_boundary_durability: bool = True
• journal_cache_events: int = 50000
• health_port: Optional[int] = None
• health_host: str = '127.0.0.1'

---

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.

WorkflowClient(store: Store, *, poll_interval: float = 0.1, admit: Optional[Callable[[str], None]] = None)

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, allow_unregistered: bool = True, start_at: Optional[datetime] = None, delay: float | timedelta | None = 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. With allow_unregistered=False a name this process has not imported raises ConfigurationError.

start_at (a timezone-aware datetime) or delay (seconds or a timedelta from now, on this client’s clock) holds the body back until that time. Give at most one; a negative, NaN, infinite or boolean delay, a naive start_at and a time more than 365 days out raise ValueError. A time already past, or a zero delay, starts the run now. The wait is the run’s first sleep: it costs no worker slot or concurrency slot, the run reads as sleeping, and a worker restart does not lose it. The body starts late by the time it takes a worker to claim the run, which under a Rate limit can be about the queue ahead of it divided by the rate; claim happens at the first attempt, so a bulk of delayed submits queues behind each other once before it waits.

Before it starts, cancel and terminate end the run without running the body, signals are buffered, and an update raises RunNotStartedError. A retry does not inherit the delay, and a continue_as_new successor starts at once.

With idempotency_key the timing counts as part of the request, as spelled: delay=60 twice returns the first run, while delay=60 after delay=120 raises IdempotencyConflict. The key lives for its usual 24 hours plus the delay.

handle

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

A handle to an existing run.

plan_retry

plan_retry(self, name: str, run_id: str) -> RetryPlan

What retrying a failed run would carry over and re-run, without starting anything.

Raises RetryRefused when the run cannot be retried and RunNotFoundError when it does not exist.

find_retry

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

The retry already started for a failed run, or None. Does not need the failed run itself, so it still answers after that run has expired.

retry

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

Start a new run that carries on from where the failed run run_id stopped.

The new run has a deterministic id, so retrying again returns the same run. It serves every step the failed run recorded before the failure instead of running it again, and runs the failed step and everything after live. The failed run is not changed. A step that started without a recorded outcome runs again; the ids of those steps are logged once as a warning when the retry is queued. A newer deployed version takes the retry over only if it replays the recorded history. Raises RetryRefused for a run that cannot be retried, RunNotFoundError for an unknown run and DuplicateRunError if the retry’s id belongs to some other 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 completed or failed, and also once a cancel has been requested, even while the run is still waiting to run its cleanup.

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, *, reason: Optional[str] = None, requested_by: Optional[str] = None) -> None

Request cancellation and queue parked runs for normal terminal cleanup.

A running step finishes before the next live boundary stops the run. Cleanup uses normal queue admission, so a worker and available capacity are needed before the cancelled run:error can be written; with none, use terminate. reason and requested_by ride on the run:cancel audit event and the cancelled terminal’s detail; only the first request records them.

terminate

terminate(self, name: str, run_id: str, *, reason: Optional[str] = None, requested_by: Optional[str] = None) -> TerminateResult

End the run now, with no worker involved.

The store writes the terminal run:error (code terminated, with reason and requested_by in its detail) and deletes the run’s lock in one step. No further step runs and finally blocks and compensations are skipped. A worker still executing the run is fenced: its writes are refused and it stands down at its next heartbeat, but a step already running may finish its side effects, and its result is discarded. The run ends the way cancel cannot guarantee: with no worker, or a hung step.

The run is ended first, so it cannot start more children, and then each open child’s parent_close policy is applied (a cancel child is terminated too, abandon children are left running), grandchildren included. A child whose submit has not landed yet starts cancelled when it does. A repeat on a run already terminated finishes a cascade an earlier call did not, so it is safe to retry after a failure; on a run that ended any other way it writes nothing.

Raises RunNotFoundError for an unknown run and TerminateUnsupported when the store lacks the terminate capability.

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

---

WorkflowSpec

WorkflowSpec(name: str, timeout: float = 300.0, max_event_count_per_run: Optional[int] = 50000, max_event_payload_bytes: PayloadCap = DEFAULT_CAP, 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)

Fields:

• name: str
• timeout: float = 300.0
• max_event_count_per_run: Optional[int] = 50000
• max_event_payload_bytes: PayloadCap = DEFAULT_CAP
• 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. With cloud=True (default) and PYRULA_CLOUD_API_KEY set, the worker reports to Pyrula Cloud and takes cloud-submitted runs; with no key it builds nothing and sends nothing. An explicit reporter replaces the cloud reporter.

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.

WorkflowWorker(*, workflows: Optional[list[Callable[..., Any]]] = None, store: Store, worker_id: Optional[str] = None, options: Optional[WorkerOptions] = None, reporter: Optional[RunReporter] = None, cloud: bool = True)

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'
• WorkflowContext = BaseContext

Functions

chain_link

chain_link(init_payload: Optional[Mapping[str, Any]], name: str, run_id: str) -> ChainLink

The chain position a run’s run:init records: a continuation successor (or a run a worker handed on to gate it) carries its root and hop in provenance, any other run is its own root.

current_run_state

current_run_state() -> Optional[RunState]

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

engine_headers

engine_headers(kind: str) -> dict[str, str]

The headers the engine puts on a store-built event of kind.

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

open_store

open_store(url: Optional[str] = None, *, default: Optional[str] = None, allow_memory: bool = False, **options: Any) -> Any

Open the workflow store url names; when None, the one PYRULA_STORE_URL names, else default.

options go to the store’s constructor; one it does not take is refused by name, so a Valkey-only option such as namespace never silently does nothing on a SQLite file.

reconstruct_state

reconstruct_state(events: list[RunEvent], *, state: Optional[ReplayState] = None, extra_arms: Optional[dict[str, ArmHandler]] = None, extra_kinds: Iterable[str] = (), extra_features: Iterable[str] = ()) -> ReplayState

Reconstruct replay state from events.

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

Raises JournalNeedsNewerRuntime, before folding anything, if the journal holds a fact this runtime cannot read (see check_journal_readable).

reopened_headers

reopened_headers(opening: RunEvent) -> dict[str, str]

The headers a reopened copy of opening carries.

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) -> tuple[str, str, Optional[RunEvent]]

Resolve (name, run_id) to the live tail of its continue_as_new chain.

Returns (tail_name, tail_run_id, tail_last_event); tail_last_event is None if the tail has no events yet. Any id in a chain resolves, so a caller keeps the id it was handed and calls this on every poll: from the root it is one tail-record read plus one last-event read, from the middle one more pair to find the root’s record. The chain has no length limit, because the tail record, not a walk over the hops, is what bounds the cost. That holds while the record lives: once it expires (a finished chain, or an ArchivingStore chain whose live copies are gone), a resolve from the root costs one last-event read per hop. A hop whose stream is gone but whose seal is still kept is followed through that seal, so a retired hop’s id keeps resolving for as long as its handle does. Raises ContinuationChainError only for a malformed chain: a run:continued payload or seal that does not parse, or a chain_hop that does not increase.

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, suspend_on: tuple[type[BaseException], ...] = ()) -> 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 (an InterruptResponse chooses the action, anything else is a “respond” value). A handler that raises or answers with an invalid action fails the run (interrupt_handler_failed) and the exception propagates, unless it is a DurableInterrupt or an instance of suspend_on: those leave the run suspended for the adapter’s own resume.

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.

stamped_engine_headers

stamped_engine_headers(headers: Mapping[str, str]) -> Iterator[None]

Make engine_headers return headers for every kind inside the block.

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] = 50000, max_event_payload_bytes: PayloadCap = DEFAULT_CAP, max_attempt_wall_seconds: Optional[float] = None, profile: Optional[WorkflowProfile] = None, version: Optional[str] = None, version_behavior: Optional[VersionBehavior] = None, 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.

The body must be async def; a plain function raises TypeError here rather than failing when a run starts. Registering a different function under an existing name emits pyrula.workflows.workflow.DuplicateWorkflowWarning.

version_behavior defaults to auto_upgrade when version is not given (the version is then a hash of the body, so every edit changes it) and to pinned when it is.

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_event_count_per_run defaults to 50,000; None sets no workflow-level cap, but a store that trims a run’s stream (ValkeyStore) still lowers the cap to what it keeps, whatever is declared. A run over the cap ends with stream:overflow and run:error rather than losing the start of its history, and ctx.continue_as_new hands the work to a fresh run. max_event_payload_bytes limits one event’s JSON payload (a step result, a submitted params) and defaults to 1,000,000 bytes; an event over it ends the run with PayloadTooLarge naming the step, and a submit over it is refused. Pass None to lift that limit.

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 unless Cloud reports a development or staging environment (then io), 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.