# Workflows > Durable workflow engine: decorators, context, stores, and the Kafka runtime. Source: https://docs.pyrula.io/reference/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. ```python 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` ```python capabilities(self) -> frozenset[str] ``` ### `close` ```python close(self) -> None ``` Close the archive, then the live store, each only if it has a `close`. ### `append_event` ```python append_event(self, name: str, run_id: str, event: RunEvent, *, fence: Optional[str] = None) -> None ``` ### `append_event_sequenced` ```python append_event_sequenced(self, name: str, run_id: str, event: RunEvent, *, retain: bool = True, fence: Optional[str] = None) -> int ``` ### `append_events_sequenced` ```python append_events_sequenced(self, name: str, run_id: str, events: list[RunEvent], *, retain: bool = True, fence: Optional[str] = None) -> list[int] ``` ### `append_events` ```python append_events(self, name: str, run_id: str, events: list[RunEvent], *, fence: Optional[str] = None) -> None ``` ### `sync_run` ```python 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` ```python 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` ```python register_projector(self, projector: Projector) -> None ``` ### `reproject_run` ```python 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` ```python 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` ```python 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` ```python load_run(self, name: str, run_id: str, *, max_events: Optional[int] = None) -> list[RunEvent] ``` ### `replay_run` ```python replay_run(self, name: str, run_id: str, *, cursor: Optional[Any] = None, count: Optional[int] = None) -> tuple[list[RunEvent], Optional[Any]] ``` ### `get_last_event` ```python get_last_event(self, name: str, run_id: str) -> Optional[RunEvent] ``` ### `read_entries_from` ```python read_entries_from(self, name: str, run_id: str, cursor: Optional[str], count: int) -> tuple[list[tuple[str, RunEvent]], Optional[str]] ``` ### `read_events_from` ```python read_events_from(self, name: str, run_id: str, from_seq: int, *, limit: Optional[int] = None) -> list[RunEvent] ``` ### `get_run_owner` ```python get_run_owner(self, name: str, run_id: str) -> Optional[str] ``` ### `list_names` ```python list_names(self) -> list[str] ``` ### `apply_retention` ```python apply_retention(self, name: str, run_id: str, ttl: Optional[int] = None) -> None ``` ### `peek_signal_or_arm` ```python 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` ```python pop_signal_match(self, name: str, run_id: str, signal_name: str, signal_id: str) -> bool ``` ### `cache_get` ```python cache_get(self, key: str) -> Optional[Any] ``` ### `cache_set` ```python cache_set(self, key: str, value: Any, ttl: float) -> None ``` ### `claim_next_run` ```python claim_next_run(self, name: str, worker_id: str, worker_version: Optional[str] = None, *, prefer: Sequence[int] = ()) -> Optional[str] ``` ### `claim_run` ```python claim_run(self, name: str, run_id: str, worker_id: str) -> None ``` ### `complete_run` ```python complete_run(self, name: str, run_id: str) -> None ``` ### `deliver_update` ```python 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` ```python peek_pending_updates(self, name: str, run_id: str) -> list[Any] ``` ### `ack_pending_update` ```python ack_pending_update(self, name: str, run_id: str, update_id: str) -> bool ``` ### `get_chain_tail` ```python get_chain_tail(self, root_name: str, root_run_id: str) -> Optional[ChainLink] ``` ### `advance_chain_tail` ```python advance_chain_tail(self, link: ChainLink) -> None ``` ### `close_chain` ```python close_chain(self, link: ChainLink) -> None ``` ### `seal_inbox` ```python seal_inbox(self, name: str, run_id: str, successor: ChainLink) -> InboxHandoff ``` ### `import_inbox` ```python import_inbox(self, link: ChainLink, handoff: InboxHandoff) -> None ``` ### `chain_adopted` ```python chain_adopted(self, name: str, run_id: str) -> bool ``` ### `read_inbox_seal` ```python read_inbox_seal(self, name: str, run_id: str) -> Optional[ChainLink] ``` ### `expire_run` ```python expire_run(self, name: str, run_id: str, ttl_seconds: int = 360) -> bool ``` ### `get_cancel_requested_at` ```python get_cancel_requested_at(self, name: str, run_id: str) -> Optional[str] ``` ### `get_idempotency_run` ```python get_idempotency_run(self, name: str, key: str) -> Optional[str] ``` ### `get_idempotency_body_hash` ```python get_idempotency_body_hash(self, name: str, key: str) -> Optional[str] ``` ### `get_interrupt_response` ```python get_interrupt_response(self, name: str, run_id: str) -> Optional[dict[str, Any]] ``` ### `get_pending_count` ```python get_pending_count(self, name: str) -> int ``` ### `get_sleeping_runs` ```python get_sleeping_runs(self, before: float, limit: int = 100) -> list[tuple[str, str]] ``` ### `heartbeat` ```python heartbeat(self, name: str, run_id: str, worker_id: str) -> tuple[Optional[str], Optional[str]] ``` ### `lock_holder` ```python lock_holder(self, name: str, run_id: str) -> Optional[str] ``` ### `kv_delete` ```python kv_delete(self, namespace: str, key: str) -> None ``` ### `kv_get` ```python kv_get(self, namespace: str, key: str) -> Optional[dict[str, Any]] ``` ### `kv_list` ```python kv_list(self, namespace: str, prefix: str = '', limit: int = 100) -> list[dict[str, Any]] ``` ### `kv_put` ```python kv_put(self, namespace: str, key: str, value: dict[str, Any]) -> None ``` ### `kv_search` ```python kv_search(self, namespace: str, query: str, limit: int = 10) -> list[dict[str, Any]] ``` ### `list_runs` ```python 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` ```python ping(self) -> None ``` ### `read_update_result` ```python read_update_result(self, name: str, run_id: str, update_id: str) -> Optional[dict[str, Any]] ``` ### `recover_pending_runs` ```python recover_pending_runs(self, name: str, now: Optional[float] = None) -> list[str] ``` ### `release_run` ```python release_run(self, name: str, run_id: str, worker_id: Optional[str] = None) -> None ``` ### `request_cancel` ```python request_cancel(self, name: str, run_id: str, *, detail: Optional[dict[str, Any]] = None) -> str ``` ### `request_pause` ```python request_pause(self, name: str, run_id: str) -> str ``` ### `terminate_run` ```python 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` ```python requeue_run(self, name: str, run_id: str, worker_id: Optional[str] = None, *, reason: str = 'owner_gone') -> None ``` ### `resume_paused_run` ```python resume_paused_run(self, name: str, run_id: str, *, wait_id: str, idempotency_key: Optional[str] = None) -> WaitOutcome ``` ### `resume_run` ```python resume_run(self, name: str, run_id: str, action: str, value: Any, *, wait_id: str, idempotency_key: Optional[str] = None) -> WaitOutcome ``` ### `send_signal` ```python 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` ```python set_idempotency_run(self, name: str, key: str, run_id: str, ttl: int = 86400) -> bool ``` ### `set_interrupted` ```python set_interrupted(self, name: str, run_id: str, event_id: str, payload: dict[str, Any]) -> None ``` ### `set_run_owner` ```python set_run_owner(self, name: str, run_id: str, owner: str) -> None ``` ### `set_sleeping` ```python set_sleeping(self, name: str, run_id: str, wake_time: float) -> None ``` ### `submit_run` ```python 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` ```python wake_run(self, name: str, run_id: str, *, wait_id: str, idempotency_key: Optional[str] = None, lease: Optional[str] = None) -> WaitOutcome ``` ### `wake_due` ```python wake_due(self, name: str, run_id: str, *, due_by: float) -> WaitOutcome ``` ### `time_out_wait` ```python time_out_wait(self, name: str, run_id: str, *, wait_id: str, idempotency_key: Optional[str] = None, lease: Optional[str] = None) -> WaitOutcome ``` ### `reopen_wait` ```python reopen_wait(self, name: str, run_id: str, *, wait_id: str, kind: str, lease: Optional[str] = None) -> WaitOutcome ``` ### `fail_interrupt` ```python fail_interrupt(self, name: str, run_id: str, *, wait_id: str, code: str, detail: dict[str, Any]) -> WaitOutcome ``` ### `write_update_result` ```python write_update_result(self, name: str, run_id: str, update_id: str, outcome: dict[str, Any]) -> None ``` ### `settle_update` ```python 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` ```python adopt_run(self, name: str, run_id: str, new_version: str, worker_id: Optional[str] = None) -> bool ``` ### `get_pause_requested_at` ```python get_pause_requested_at(self, name: str, run_id: str) -> Optional[str] ``` ### `get_run_controls` ```python get_run_controls(self, name: str, run_id: str) -> tuple[Optional[str], Optional[str]] ``` ### `get_run_version` ```python get_run_version(self, name: str, run_id: str) -> Optional[tuple[str, str]] ``` ### `get_run_rank` ```python get_run_rank(self, name: str, run_id: str) -> Optional[int] ``` ### `note_replay_check_timeout` ```python note_replay_check_timeout(self, name: str, run_id: str, worker_id: Optional[str] = None) -> int ``` ### `hand_back_run` ```python hand_back_run(self, name: str, run_id: str, worker_id: Optional[str] = None) -> None ``` ### `register_version` ```python register_version(self, name: str, version: str) -> int ``` ### `version_rank` ```python version_rank(self, name: str, version: str) -> Optional[int] ``` ### `get_store_binding` ```python get_store_binding(self) -> Optional[StoreBinding] ``` ### `bind_store` ```python bind_store(self, binding: StoreBinding) -> StoreBinding ``` ### `clear_store_binding` ```python clear_store_binding(self, expected: StoreBinding) -> bool ``` ### `submitted_at` ```python submitted_at(self, name: str, run_id: str) -> Optional[float] ``` --- ## `Batch` A batch of records from a source. ```python 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` ```python cache_get(self, key: str) -> Optional[Any] ``` ### `cache_set` ```python cache_set(self, key: str, value: Any, ttl: float) -> None ``` --- ## `ChainLink` ```python 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` ```python root(cls, name: str, run_id: str) -> Self ``` ### `successor` ```python successor(self, name: str, run_id: str) -> ChainLink ``` ### `provenance_source` ```python provenance_source(self) -> dict[str, Any] ``` ### `continued_payload` ```python continued_payload(self) -> dict[str, Any] ``` The ``run:continued`` payload that hands off to *this* link. ### `from_continued_payload` ```python 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` ```python to_json(self) -> str ``` ### `from_json` ```python 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` ```python 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` ```python advance_chain_tail(self, link: ChainLink) -> None ``` Record ``link`` as the chain's tail unless a later hop is already recorded. ### `close_chain` ```python 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` ```python 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` ```python 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` ```python chain_adopted(self, name: str, run_id: str) -> bool ``` Whether this successor has already imported its predecessor's inbox. ### `read_inbox_seal` ```python 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": , "events": [...], "count": }``. Named ``Collect`` (not ``Batch``) to avoid clashing with the streaming ``Batch`` type. ```python 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` ```python 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. ```python 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` ```python capabilities(self) -> frozenset[str] ``` ### `submit_run` ```python 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` ```python claim_next_run(self, name: str, worker_id: str, worker_version: Optional[str] = None, *, prefer: Sequence[int] = ()) -> Optional[str] ``` ### `claim_run` ```python claim_run(self, name: str, run_id: str, worker_id: str) -> None ``` ### `release_run` ```python release_run(self, name: str, run_id: str, worker_id: Optional[str] = None) -> None ``` ### `requeue_run` ```python requeue_run(self, name: str, run_id: str, worker_id: Optional[str] = None, *, reason: str = 'owner_gone') -> None ``` ### `recover_pending_runs` ```python recover_pending_runs(self, name: str, now: Optional[float] = None) -> list[str] ``` ### `get_pending_count` ```python get_pending_count(self, name: str) -> int ``` ### `heartbeat` ```python heartbeat(self, name: str, run_id: str, worker_id: str) -> tuple[Optional[str], Optional[str]] ``` ### `request_cancel` ```python request_cancel(self, name: str, run_id: str, *, detail: Optional[dict[str, Any]] = None) -> str ``` ### `get_cancel_requested_at` ```python get_cancel_requested_at(self, name: str, run_id: str) -> Optional[str] ``` ### `request_pause` ```python request_pause(self, name: str, run_id: str) -> str ``` ### `get_pause_requested_at` ```python get_pause_requested_at(self, name: str, run_id: str) -> Optional[str] ``` ### `get_run_controls` ```python get_run_controls(self, name: str, run_id: str) -> tuple[Optional[str], Optional[str]] ``` ### `resume_paused_run` ```python resume_paused_run(self, name: str, run_id: str, *, wait_id: str, idempotency_key: Optional[str] = None) -> WaitOutcome ``` ### `expire_run` ```python expire_run(self, name: str, run_id: str, ttl_seconds: int = 360) -> bool ``` ### `lock_holder` ```python lock_holder(self, name: str, run_id: str) -> Optional[str] ``` ### `append_event` ```python append_event(self, name: str, run_id: str, event: RunEvent, *, fence: Optional[str] = None) -> None ``` ### `append_events` ```python append_events(self, name: str, run_id: str, events: list[RunEvent], *, fence: Optional[str] = None) -> None ``` ### `replay_run` ```python replay_run(self, name: str, run_id: str, **kw: Any) -> tuple[list[RunEvent], Optional[Any]] ``` ### `load_run` ```python load_run(self, name: str, run_id: str, **kw: Any) -> list[RunEvent] ``` ### `get_last_event` ```python get_last_event(self, name: str, run_id: str) -> Optional[RunEvent] ``` ### `read_entries_from` ```python read_entries_from(self, name: str, run_id: str, cursor: Optional[str], count: int) -> tuple[list[tuple[str, RunEvent]], Optional[str]] ``` ### `set_run_owner` ```python set_run_owner(self, name: str, run_id: str, owner: str) -> None ``` ### `get_run_owner` ```python get_run_owner(self, name: str, run_id: str) -> Optional[str] ``` ### `get_store_binding` ```python get_store_binding(self) -> Optional[StoreBinding] ``` ### `bind_store` ```python bind_store(self, binding: StoreBinding) -> StoreBinding ``` ### `clear_store_binding` ```python clear_store_binding(self, expected: StoreBinding) -> bool ``` ### `ping` ```python ping(self) -> None ``` ### `list_runs` ```python 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` ```python set_interrupted(self, name: str, run_id: str, event_id: str, payload: dict[str, Any]) -> None ``` ### `resume_run` ```python resume_run(self, name: str, run_id: str, action: str, value: Any, *, wait_id: str, idempotency_key: Optional[str] = None) -> WaitOutcome ``` ### `fail_interrupt` ```python fail_interrupt(self, name: str, run_id: str, *, wait_id: str, code: str, detail: dict[str, Any]) -> WaitOutcome ``` ### `hand_back_run` ```python hand_back_run(self, name: str, run_id: str, worker_id: Optional[str] = None) -> None ``` ### `terminate_run` ```python 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` ```python get_run_version(self, name: str, run_id: str) -> Optional[tuple[str, str]] ``` ### `adopt_run` ```python adopt_run(self, name: str, run_id: str, new_version: str, worker_id: Optional[str] = None) -> bool ``` ### `note_replay_check_timeout` ```python note_replay_check_timeout(self, name: str, run_id: str, worker_id: Optional[str] = None) -> int ``` ### `register_version` ```python register_version(self, name: str, version: str) -> int ``` ### `version_rank` ```python version_rank(self, name: str, version: str) -> Optional[int] ``` ### `get_run_rank` ```python get_run_rank(self, name: str, run_id: str) -> Optional[int] ``` ### `complete_run` ```python complete_run(self, name: str, run_id: str) -> None ``` ### `get_interrupt_response` ```python get_interrupt_response(self, name: str, run_id: str) -> Optional[dict[str, Any]] ``` ### `get_idempotency_run` ```python get_idempotency_run(self, name: str, key: str) -> Optional[str] ``` ### `set_idempotency_run` ```python set_idempotency_run(self, name: str, key: str, run_id: str, ttl: int = 86400) -> bool ``` ### `get_idempotency_body_hash` ```python get_idempotency_body_hash(self, name: str, key: str) -> Optional[str] ``` ### `cache_get` ```python cache_get(self, key: str) -> Optional[Any] ``` ### `cache_set` ```python cache_set(self, key: str, value: Any, ttl: float) -> None ``` ### `set_sleeping` ```python set_sleeping(self, name: str, run_id: str, wake_time: float) -> None ``` ### `get_sleeping_runs` ```python get_sleeping_runs(self, before: float, limit: int = 100) -> list[tuple[str, str]] ``` ### `wake_run` ```python wake_run(self, name: str, run_id: str, *, wait_id: str, idempotency_key: Optional[str] = None, lease: Optional[str] = None) -> WaitOutcome ``` ### `wake_due` ```python wake_due(self, name: str, run_id: str, *, due_by: float) -> WaitOutcome ``` ### `time_out_wait` ```python time_out_wait(self, name: str, run_id: str, *, wait_id: str, idempotency_key: Optional[str] = None, lease: Optional[str] = None) -> WaitOutcome ``` ### `reopen_wait` ```python reopen_wait(self, name: str, run_id: str, *, wait_id: str, kind: str, lease: Optional[str] = None) -> WaitOutcome ``` ### `kv_get` ```python kv_get(self, namespace: str, key: str) -> Optional[dict[str, Any]] ``` ### `kv_put` ```python kv_put(self, namespace: str, key: str, value: dict[str, Any]) -> None ``` ### `kv_delete` ```python kv_delete(self, namespace: str, key: str) -> None ``` ### `kv_list` ```python kv_list(self, namespace: str, prefix: str = '', limit: int = 100) -> list[dict[str, Any]] ``` ### `kv_search` ```python kv_search(self, namespace: str, query: str, limit: int = 10) -> list[dict[str, Any]] ``` ### `send_signal` ```python 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` ```python 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` ```python pop_signal_match(self, name: str, run_id: str, signal_name: str, signal_id: str) -> bool ``` ### `deliver_update` ```python 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` ```python peek_pending_updates(self, name: str, run_id: str) -> list[Any] ``` ### `ack_pending_update` ```python ack_pending_update(self, name: str, run_id: str, update_id: str) -> bool ``` ### `write_update_result` ```python write_update_result(self, name: str, run_id: str, update_id: str, outcome: dict[str, Any]) -> None ``` ### `settle_update` ```python 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` ```python read_update_result(self, name: str, run_id: str, update_id: str) -> Optional[dict[str, Any]] ``` ### `get_chain_tail` ```python get_chain_tail(self, root_name: str, root_run_id: str) -> Optional[ChainLink] ``` ### `advance_chain_tail` ```python advance_chain_tail(self, link: ChainLink) -> None ``` ### `close_chain` ```python close_chain(self, link: ChainLink) -> None ``` ### `seal_inbox` ```python seal_inbox(self, name: str, run_id: str, successor: ChainLink) -> InboxHandoff ``` ### `import_inbox` ```python import_inbox(self, link: ChainLink, handoff: InboxHandoff) -> None ``` ### `chain_adopted` ```python chain_adopted(self, name: str, run_id: str) -> bool ``` ### `read_inbox_seal` ```python 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. ```python 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`. ```python Debounce(key: str | Callable[[dict], str], window_seconds: float) ``` **Fields:** - `key: str | Callable[[dict], str]` - `window_seconds: float` ### `resolve` ```python 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". ```python 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. ```python DurableInvocationCancelled(name: str, run_id: str, detail: str) ``` **Fields:** - `code = ErrorCode.CANCELLED` --- ## `DurableInvocationError` A run driven by ``run_and_await`` ended in error. ```python 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. ```python 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` ```python 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` ```python 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` ```python load_run(self, name: str, run_id: str, *, max_events: Optional[int] = None) -> list[RunEvent] ``` ### `replay_run` ```python replay_run(self, name: str, run_id: str, *, cursor: Optional[Any] = None, count: Optional[int] = None) -> tuple[list[RunEvent], Optional[Any]] ``` ### `get_last_event` ```python get_last_event(self, name: str, run_id: str) -> Optional[RunEvent] ``` ### `read_entries_from` ```python 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. ```python 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` ```python get_idempotency_run(self, name: str, key: str) -> Optional[str] ``` ### `set_idempotency_run` ```python set_idempotency_run(self, name: str, key: str, run_id: str, ttl: int = 86400) -> bool ``` ### `get_idempotency_body_hash` ```python 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. ```python InMemorySource(*, source_id: str = 'test-topic', records: Optional[list[T]] = None, batch_size: int = 10) ``` **Fields:** - `source_id: str` ### `poll` ```python poll(self, timeout_ms: int = 5000) -> Optional[Batch[T]] ``` ### `seek` ```python seek(self, source_id: str, partition: int, offset: int) -> None ``` ### `get_committed` ```python get_committed(self, source_id: str, partition: int) -> Optional[int] ``` ### `partitions` ```python partitions(self) -> list[tuple[str, int]] ``` ### `inject_committed` ```python 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. ```python 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] = , 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] = ` - `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()``. ```python 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` ```python 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` ```python 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` ```python run_next(self, name: str) -> Optional[list[RunEvent]] ``` ### `run` ```python 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` ```python 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` ```python 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. ```python JournalNeedsNewerRuntime(message: Optional[str] = None, *, missing: Sequence[str] = ()) ``` **Fields:** - `missing` --- ## `KVStore` ### `kv_get` ```python kv_get(self, namespace: str, key: str) -> Optional[dict[str, Any]] ``` ### `kv_put` ```python kv_put(self, namespace: str, key: str, value: dict[str, Any]) -> None ``` ### `kv_delete` ```python kv_delete(self, namespace: str, key: str) -> None ``` ### `kv_list` ```python kv_list(self, namespace: str, prefix: str = '', limit: int = 100) -> list[dict[str, Any]] ``` ### `kv_search` ```python 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` ```python 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` ```python 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` ```python 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` ```python 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` ```python release_run(self, name: str, run_id: str, worker_id: Optional[str] = None) -> None ``` ### `requeue_run` ```python 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` ```python recover_pending_runs(self, name: str, now: Optional[float] = None) -> list[str] ``` ### `get_pending_count` ```python get_pending_count(self, name: str) -> int ``` ### `heartbeat` ```python heartbeat(self, name: str, run_id: str, worker_id: str) -> tuple[Optional[str], Optional[str]] ``` ### `lock_holder` ```python 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` ```python 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` ```python get_cancel_requested_at(self, name: str, run_id: str) -> Optional[str] ``` ### `get_run_controls` ```python 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` ```python expire_run(self, name: str, run_id: str, ttl_seconds: int = 360) -> bool ``` ### `set_run_owner` ```python set_run_owner(self, name: str, run_id: str, owner: str) -> None ``` ### `get_run_owner` ```python get_run_owner(self, name: str, run_id: str) -> Optional[str] ``` ### `get_run_version` ```python 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` ```python 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` ```python 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` ```python 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` ```python version_rank(self, name: str, version: str) -> Optional[int] ``` The rank ``register_version`` assigned, or None for a version never seen. ### `get_run_rank` ```python 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` ```python 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` ```python 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` ```python 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` ```python bind_store(self, binding: StoreBinding) -> StoreBinding ``` Atomically bind an unbound store and return the stored winner. ### `clear_store_binding` ```python clear_store_binding(self, expected: StoreBinding) -> bool ``` Remove the binding if it equals ``expected``; return whether it was deleted. ### `submitted_at` ```python 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` ```python 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` ```python 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` ```python 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` ```python ping(self) -> None ``` ### `list_runs` ```python 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` ```python set_interrupted(self, name: str, run_id: str, event_id: str, payload: dict[str, Any]) -> None ``` ### `resume_run` ```python 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` ```python 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` ```python 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` ```python complete_run(self, name: str, run_id: str) -> None ``` ### `get_interrupt_response` ```python 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). ```python Limit(key: str | Callable[[dict], str], max: int) ``` **Fields:** - `key: str | Callable[[dict], str]` - `max: int` ### `resolve` ```python 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. ```python 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` ```python list_names(self) -> list[str] ``` --- ## `NestedDurableOperationError` Durable orchestration attempted inside an atomic step or tool. ```python 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. ```python 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. ```python PostgresArchive(*, dsn: str, replay_page_size: int = 1000, pool: Optional[Any] = None, max_connections: int = 8) ``` **Fields:** - `dsn: str` - `replay_page_size: int` ### `close` ```python close(self) -> None ``` Close the connections this archive opened itself. A pool it was handed belongs to the caller. ### `check_schema` ```python 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` ```python 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` ```python 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` ```python load_archived_run(self, name: str, run_id: str) -> list[RunEvent] ``` ### `load_archived_run_owner` ```python load_archived_run_owner(self, name: str, run_id: str) -> Optional[str] ``` ### `replay_archived_run` ```python replay_archived_run(self, name: str, run_id: str, *, cursor: Optional[Any], count: Optional[int]) -> tuple[list[RunEvent], Optional[Any]] ``` ### `read_archived_entries_from` ```python 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` ```python list_archived_names(self) -> list[str] ``` ### `list_archived_runs` ```python 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. ```python 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` ```python 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). ```python ReplayState(recorded_call_results: dict[tuple[str, int], Any] = , recorded_call_results_by_id: dict[tuple[str, str], Any] = , recorded_call_fingerprints: dict[tuple[str, int], str] = , recorded_call_fingerprints_by_id: dict[tuple[str, str], str] = , recorded_call_slots: dict[str, list[int]] = , recorded_call_failures: dict[tuple[str, int], tuple[str, int]] = , recorded_call_reservations: dict[tuple[str, int], str] = , recorded_call_questions: dict[tuple[str, int], tuple[str, int]] = , manual_checkpoint_counts: dict[str, int] = , emit_counts: dict[str, int] = , fanout_results: dict[str, list[Any]] = , fanout_fingerprints: dict[str, str] = , fanout_pending: dict[str, list[dict[str, str]]] = , fanout_pending_fingerprints: dict[str, str] = , child_results: dict[tuple[str, int], Any] = , signal_send_results: dict[str, Any] = , signal_send_fingerprints: dict[str, str] = , deterministic_results: dict[tuple[str, int], Any] = , interrupt_results: dict[tuple[str, int], dict[str, Any]] = , interrupt_fingerprints: dict[tuple[str, int], str] = , sleep_results: dict[str, set[int]] = , sleep_armed: dict[tuple[str, int], str] = , sleep_wait_ids: dict[tuple[str, int], str] = , open_sleep: Optional[tuple[str, int]] = None, signal_armed: dict[tuple[str, int], tuple[str, Optional[str]]] = , interrupt_armed: dict[tuple[str, int], tuple[str, Optional[str]]] = , signal_results: dict[tuple[str, int], dict[str, Any]] = , signal_fingerprints: dict[tuple[str, int], str] = , journaled_signal_ids: set[str] = , named_step_results: dict[tuple[str, int], Any] = , named_step_errors: dict[tuple[str, int], list[dict[str, Any]]] = , 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] = , child_fingerprints: dict[tuple[str, int], str] = , captured_step_keys: set[tuple[str, int]] = , captured_call_keys: set[tuple[str, str]] = , stream_committed_offsets: dict[tuple[str, int], int] = , stream_committed_batch_counts: dict[tuple[str, int], int] = , stream_carry: dict[str, Any] = , patch_markers: set[str] = , updates_by_frontier: dict[int, list[dict[str, Any]]] = , journaled_update_ids: set[str] = , journaled_update_outcomes: dict[str, dict[str, Any]] = , journaled_effect_count: int = 0) ``` **Fields:** - `recorded_call_results: dict[tuple[str, int], Any] = ` - `recorded_call_results_by_id: dict[tuple[str, str], Any] = ` - `recorded_call_fingerprints: dict[tuple[str, int], str] = ` - `recorded_call_fingerprints_by_id: dict[tuple[str, str], str] = ` - `recorded_call_slots: dict[str, list[int]] = ` - `recorded_call_failures: dict[tuple[str, int], tuple[str, int]] = ` - `recorded_call_reservations: dict[tuple[str, int], str] = ` - `recorded_call_questions: dict[tuple[str, int], tuple[str, int]] = ` - `manual_checkpoint_counts: dict[str, int] = ` - `emit_counts: dict[str, int] = ` - `fanout_results: dict[str, list[Any]] = ` - `fanout_fingerprints: dict[str, str] = ` - `fanout_pending: dict[str, list[dict[str, str]]] = ` - `fanout_pending_fingerprints: dict[str, str] = ` - `child_results: dict[tuple[str, int], Any] = ` - `signal_send_results: dict[str, Any] = ` - `signal_send_fingerprints: dict[str, str] = ` - `deterministic_results: dict[tuple[str, int], Any] = ` - `interrupt_results: dict[tuple[str, int], dict[str, Any]] = ` - `interrupt_fingerprints: dict[tuple[str, int], str] = ` - `sleep_results: dict[str, set[int]] = ` - `sleep_armed: dict[tuple[str, int], str] = ` - `sleep_wait_ids: dict[tuple[str, int], str] = ` - `open_sleep: Optional[tuple[str, int]] = None` - `signal_armed: dict[tuple[str, int], tuple[str, Optional[str]]] = ` - `interrupt_armed: dict[tuple[str, int], tuple[str, Optional[str]]] = ` - `signal_results: dict[tuple[str, int], dict[str, Any]] = ` - `signal_fingerprints: dict[tuple[str, int], str] = ` - `journaled_signal_ids: set[str] = ` - `named_step_results: dict[tuple[str, int], Any] = ` - `named_step_errors: dict[tuple[str, int], list[dict[str, Any]]] = ` - `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] = ` - `child_fingerprints: dict[tuple[str, int], str] = ` - `captured_step_keys: set[tuple[str, int]] = ` - `captured_call_keys: set[tuple[str, str]] = ` - `stream_committed_offsets: dict[tuple[str, int], int] = ` - `stream_committed_batch_counts: dict[tuple[str, int], int] = ` - `stream_carry: dict[str, Any] = ` - `patch_markers: set[str] = ` - `updates_by_frontier: dict[int, list[dict[str, Any]]] = ` - `journaled_update_ids: set[str] = ` - `journaled_update_outcomes: dict[str, dict[str, Any]] = ` - `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. ```python 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] = , metadata: dict[str, Any] = , 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] = ` - `metadata: dict[str, Any] = ` - `version: str = ''` ### `provenance` ```python 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. ```python 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` ```python 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` ```python 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. ```python 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. ```python 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] = ) ``` **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] = ` --- ## `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. ```python RunHandle(client: WorkflowClient, name: str, run_id: str) ``` **Fields:** - `name: str` - `run_id: str` ### `signal` ```python signal(self, signal: str, payload: Optional[dict[str, Any]] = None, *, signal_id: Optional[str] = None) -> str ``` ### `query` ```python query(self, query: str, *args: Any, **kwargs: Any) -> Any ``` ### `update` ```python update(self, update: str, payload: Optional[dict[str, Any]] = None, *, update_id: Optional[str] = None, timeout: float = 30.0) -> Any ``` ### `cancel` ```python cancel(self, *, reason: Optional[str] = None, requested_by: Optional[str] = None) -> None ``` ### `terminate` ```python terminate(self, *, reason: Optional[str] = None, requested_by: Optional[str] = None) -> TerminateResult ``` ### `retry` ```python retry(self) -> RunHandle ``` ### `resume` ```python resume(self, action: str, value: Any = None, *, wait_id: Optional[str] = None) -> None ``` ### `status` ```python status(self) -> RunStatus ``` ### `result` ```python result(self, *, timeout: Optional[float] = None, on_interrupt: Any = None) -> Any ``` ### `events` ```python 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` ```python 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. ```python 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` ```python 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` ```python 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` ```python 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` ```python ping(self) -> None ``` ### `list_names` ```python list_names(self) -> list[str] ``` ### `close` ```python close(self) -> None ``` --- ## `ScheduleBackend` ### `schedule_create` ```python schedule_create(self, schedule: Schedule) -> None ``` ### `schedule_get` ```python schedule_get(self, schedule_id: str) -> Optional[Schedule] ``` ### `schedule_list` ```python schedule_list(self, name: Optional[str] = None) -> list[Schedule] ``` ### `schedule_update` ```python schedule_update(self, schedule: Schedule) -> None ``` ### `schedule_advance` ```python schedule_advance(self, fired: Schedule, updates: dict[str, Any]) -> bool ``` ### `schedule_patch` ```python 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` ```python schedule_delete(self, schedule_id: str) -> bool ``` ### `schedule_due` ```python schedule_due(self, now: float) -> list[Schedule] ``` ### `schedule_claim_fire` ```python schedule_claim_fire(self, schedule_id: str, slot: int) -> bool ``` --- ## `SchemaVersionMismatch` An event's schema_version major is newer than the runtime understands. ```python SchemaVersionMismatch(message: Optional[str] = None, *, missing: Sequence[str] = ()) ``` --- ## `Signal` ```python Signal(name: str, payload: dict[str, Any], sent_at: str) ``` **Fields:** - `name: str` - `payload: dict[str, Any]` - `sent_at: str` --- ## `SignalStore` ### `send_signal` ```python 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` ```python 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` ```python 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` ```python poll(self, timeout_ms: int = 5000) -> Optional[Batch[T]] ``` Return the next batch, or ``None`` if no data within timeout. ### `seek` ```python seek(self, source_id: str, partition: int, offset: int) -> None ``` Seek to a specific offset (used on resume). ### `get_committed` ```python get_committed(self, source_id: str, partition: int) -> Optional[int] ``` Return the last committed offset for a partition, or None. ### `partitions` ```python partitions(self) -> list[tuple[str, int]] ``` Return all (source_id, partition) pairs. --- ## `StepContext` Context available inside a decorated step. ```python 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. ```python 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. ```python 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` ```python belongs_to(self, project_id: str, environment: str) -> bool ``` ### `to_json` ```python to_json(self) -> str ``` Canonical, so a compare-and-delete can compare stored text. ### `from_json` ```python from_json(cls, raw: str) -> Self ``` ### `now` ```python 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` ```python 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. ```python 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` ```python set_sleeping(self, name: str, run_id: str, wake_time: float) -> None ``` ### `get_sleeping_runs` ```python get_sleeping_runs(self, before: float, limit: int = 100) -> list[tuple[str, str]] ``` ### `wake_run` ```python 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` ```python 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` ```python 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` ```python 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. ```python 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` ```python from_url(cls, url: str, **kwargs: Any) -> Self ``` ### `resolve_cluster` ```python 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` ```python list_names(self) -> list[str] ``` ### `shard_of` ```python 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``. ```python 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`` ```python 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. ```python WorkflowClient(store: Store, *, poll_interval: float = 0.1, admit: Optional[Callable[[str], None]] = None) ``` **Fields:** - `store: Store` ### `submit` ```python 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` ```python handle(self, name: str, run_id: str) -> RunHandle ``` A handle to an existing run. ### `plan_retry` ```python 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` ```python 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` ```python 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` ```python 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` ```python 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` ```python 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` ```python 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` ```python 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` ```python 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` ```python 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` ```python status(self, name: str, run_id: str) -> RunStatus ``` The run's current status. ### `result` ```python 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` ```python events(self, name: str, run_id: str) -> list[RunEvent] ``` The run's own journal, oldest first (not followed across continue_as_new). --- ## `WorkflowSpec` ```python 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. ```python 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` ```python 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` ```python run(self) -> None ``` Run all supervisor loops until ``stop()`` is called. ## Constants - `EVENT_SCHEMA_VERSION = '1'` - `WorkflowContext = BaseContext` ## Functions ### `chain_link` ```python 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` ```python current_run_state() -> Optional[RunState] ``` The RunState of the currently executing run body, or None if not inside a run. ### `engine_headers` ```python engine_headers(kind: str) -> dict[str, str] ``` The headers the engine puts on a store-built event of ``kind``. ### `make_archiving_postgres` ```python 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` ```python 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` ```python 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` ```python reopened_headers(opening: RunEvent) -> dict[str, str] ``` The headers a reopened copy of ``opening`` carries. ### `replay_all_events` ```python 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` ```python 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` ```python 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` ```python stamped_engine_headers(headers: Mapping[str, str]) -> Iterator[None] ``` Make ``engine_headers`` return ``headers`` for every kind inside the block. ### `step` ```python 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` ```python 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.