Workflows and agents
that survive crashes

Pyrula writes each step of a run to a journal as it happens. When a worker dies, another one replays the journal and continues from the first unfinished step. Durable workflows and LLM agents for Python, on a Rust core.

Read the quickstartSource on GitHub →
$pip install "pyrula[all]"
● nightly_orders.pyrun_01KA3C · journal · live
from pyrula.workflows import workflow @workflow async def nightly_orders(ctx, day: str): raw = await ctx.step("extract", export_orders, day) clean = await ctx.step("clean", drop_test_and_dupes, raw) usd = await ctx.step("to_usd", convert_to_usd, clean) return await ctx.step("load", merge_into_warehouse, usd, day)
0001run:initnightly_ordersworker-2
0002step:pendingextractworker-2
0003step:doneextract → 2.1M rows14m 02s
0004step:pendingcleanworker-2
0005step:doneclean → 1.98M rows3m 40s
0006step:pendingto_usdworker-2
0007step:doneto_usd → 1.98M rows2m 11s
0008step:pendingloadworker-2
worker-2 is killed mid-load. worker-5 claims the run and replays 0001–0008; extract, clean and to_usd return their recorded results.
0009step:pendingloadworker-5
0010step:doneload → 1.98M merged4m 05s
0011run:completenightly_orders26m 12s

A workflow and the journal it writes.

Workflows

pyrula-workflowsReplay and recovery →

Recovery

Before a step runs, the worker writes step:pending. After it succeeds, step:done. Recovery resumes from the first step without a step:done. Sleeps, timers and signals are recorded too, so a waiting run picks up on another worker.

nightly_orders journal
seqstepstatusworkertime
0003extractdoneworker-214m 02s
0005cleandoneworker-23m 40s
0007to_usddoneworker-22m 11s
0010loaddoneworker-54m 05s
0011runcompleteworker-526m 12s

Sleeps and signals

ctx.sleep suspends a run until a wake time that Pyrula records. ctx.wait_for_signal suspends it until a signal with that name arrives, and a signal sent before the run reaches the wait is buffered.

Timers →Signals →
await ctx.sleep(timedelta(days=3))await ctx.wait_for_signal("payment_confirmed")invoice_followup
send_invoicestep:donesend_reminderstep:pendingmark_paidstep:pendingrun:sleepingrun:wokenrun:signal_waitingrun:signal_receivedworker-2worker-5worker-3send_invoicestep:donesend_reminderstep:pendingmark_paidstep:pendingrun:sleepingrun:wokenrun:signal_waitingrun:signal_receivedworker-2worker-5worker-3

What-if runs

Pick a finished run and a point in it, change an input, a step's result, the model or the system prompt, and one of your workers runs the rest. Calls up to the fork are replayed from the recording, and anything that could touch the outside world is stopped unless you allow it. The result is a new run, linked to the one it came from.

Here the threshold changes after scoring, so all 128 model calls come from the recording.

"source": {"runId": "run_01K9F8", "forkAfterSeq": 4},
"overrides": {"inputsPatch": {"threshold": 0.75}}

replayed from the recording · 128 model calls, not paid twiceclick a step to fork therefetch_invoices128 rowsseq 1–2score_risk@agent · 128 callsseq 3–4flagthreshold 0.90seq 5–6reportpdfseq 7–83 flagged of 128threshold 0.90forked here, one input changedflagthreshold 0.75ran livereportpdfran live11 flagged of 128what-if · after seq 4replayed from therecording · 128 calls,not paid twiceclick a step to fork therefetch_invoices128 rowsseq 1–2score_risk128 model callsseq 3–4flagthreshold 0.90seq 5–6reportpdfseq 7–83 flagged of 128threshold 0.90forked hereflagthreshold 0.75ran livereportpdfran live11 flagged of 128what-if · after seq 4

Agents

pyrula-agentsAgents →

Agent turns

Every turn is a workflow underneath. Finished model and tool calls are journaled, so a crashed turn resumes without repeating paid calls. Serve it over HTTP or drive it from Kafka. A client that drops reconnects from its last event id and misses nothing.

support_triage · turn 7 · worker-3› customer says the refund never arrived, order F-100tool:result lookup_order · tool:result refund_status · llm:start · llm:doneLooking up F-100. The charge step completed on 10/28 and the refund is waiting on a bank callback. I'll keep the ticket open and check the callback again at 3pm.

Kafka

pyrula-kafkaKafka →

Event-driven workflows and agents

@kafka_workflow starts a durable run for each record on an input topic and produces the result to an output topic. @kafka_agent does the same for an agent. The Kafka client underneath is pure Rust, with no librdkafka.

The default is at-least-once delivery. With delivery="exactly_once", the output record and the input offset are committed in one transaction. Delivery guarantees →

score.pypyrula-workflows
from pyrula.workflows.kafka import kafka_workflow
# Invoice and Scored are pydantic models
@kafka_workflow(
    input_topic="invoices",
    group="scorer",
    output_topic="scored-invoices",
    input_schema=Invoice,
    output_schema=Scored,
    delivery="exactly_once",
)
async def score(ctx, invoice: Invoice) -> Scored:
    risk = await ctx.step(
        "score", model_score, invoice)
    return Scored(id=invoice.id, risk=risk)

Pyrula Cloud

pyrula-cloud-clientConnect a worker →

Run timeline

A hosted UI for watching and controlling runs. Runs from every connected worker appear in one timeline with their state: running, sleeping, waiting for a signal or complete. From a finished run, Cloud starts a what-if run.

Connecting is one environment variable, PYRULA_CLOUD_API_KEY. The code doesn't change.

Open Pyrula Cloud →

Python library. Rust core. Scala inspired.

Workflows and agents are plain Python functions. The core types, serde and the Kafka client are written in Rust and loaded through PyO3.

The API follows Scala's standard library. Either, Option and Try compose with map and flat_map, and the collections are immutable like Scala's List and Map. Every Pyrula package depends on pyrula, and you can use it on its own.

Packages

  1. pyrulaRust-backed core: functional types and serde.Primitives →
  2. pyrula-kafkaPure-Rust, batch-first Kafka client.Kafka →
  3. pyrula-workflowsDurable execution engine on the Rust-backed core.Workflows →
  4. pyrula-agentsDurable agent runtime built on pyrula.workflows.Agents →
  5. pyrula-cloud-clientConnector for Pyrula Cloud.Cloud →

Primitives

from pyrula import Either, IList, Option def qty(raw: str) -> Either[int, str]: if raw.isdigit(): return Either.ok(int(raw)) return Either.err(f"bad qty: {raw}") rows = IList(["3", "12", "x"]).map(qty) rows.collect_err() # IList(['bad qty: x']) nums = rows.collect_ok() # IList([3, 12]) big: Option[int] = nums.find(lambda n: n > 9) big.get_or_else(0) # 12
  • EitherHolds a value or a typed error, without exceptions.
  • ValidatedLike Either, but collects every error instead of stopping at the first.
  • OptionPresent or absent, with no None ambiguity.
  • TryCaptures a result or the exception that broke it.
  • IList · IMap · ISetImmutable and structurally shared, so copies are cheap.