Runs as Event Streams

A run is one append-only event stream in the store. Workers replay it to recover and clients read it to follow the run live. Any worker rebuilds state by reading from the start; there are no snapshots, and the stream wins over any secondary index.

Events

Events are RunEvent records tagged with an EventKind.

from pyrula.workflows import RunEvent, EventKind, RunStatus

Group	Kinds
Lifecycle	run:init, run:complete, run:error, run:cancel, run:continued, run:requeued, run:interrupted, run:resumed, run:sleeping, run:woken, run:pause_requested, run:paused, run:unpaused, run:signal_waiting, run:signal_received, run:wait_timed_out
Updates	run:update_received, run:update_completed, run:update_failed
Step	step:pending, step:done, step:error
Progress	run:status (from ctx.emit_status(text)), plus whatever you ctx.emit
Internal	heartbeat, stream:overflow, patch_marker

ctx.emit raises ValueError for framework kinds: the prefixes run:, step:, llm:, block:, tool:, agent:, context:, stream:, child:, select:, mcp:, saga:, approval:, and the bare heartbeat and patch_marker. Use your own prefix such as app:.

run:complete.duration_ms is the wall-clock time of the attempt that finished; a wait before a resume is not counted. RunStatus is derived from the stream: pending, in_progress, complete, cancelled, terminated, error, quarantined, interrupted, paused, sleeping, signal_waiting.

@workflow(name="onboard_user")
async def onboard_user(ctx, email: str) -> str:
    account_id = await ctx.step("create_account", lambda: create_account(email))
    await ctx.step("send_welcome", lambda: send_welcome_email(account_id))
    return account_id


store = MemoryStore()
runner = InlineWorkflowRunner(store=store)
run = await runner.client.submit("onboard_user", {"email": "ada@example.com"})
await run.result()

for event in await run.events():
    print(event.kind, event.payload.get("result"))
# run:init       None
# step:pending   None
# step:done      acct_42          <- create_account result
# step:pending   None
# step:done      None             <- send_welcome result
# run:complete   acct_42

The event envelope

Every event has six fields: kind, payload, run_id, schema_version, run_seq, headers. The set is closed; new facts ride in headers, a string mapping the framework sets. Readers ignore header names they don’t know. Reserved names: requires (feature strings a reader must know to replay; written on start-at and case-override runs), encoding, ignorable.

A terminal event (run:complete, run:error, run:continued) is the last event of the attempt that wrote it: the worker drops any durable event a still-running step adds after it, so neither the journal nor the events the attempt returns go past it (a streamed event, which is not journaled, can still follow in the returned list). A store refuses a batch that puts an event behind a terminal, whole, by raising EventAfterTerminal (a LockLost subclass, so a plain LockLost still means the lease is lost); EventLogStoreContract checks that too. An attempt whose write a store refused ends its run with a run:error coded event_refused, because the same events would be refused again; if the store already shows the run as ended (a sibling’s late write), the attempt only stands down. A store must accept a lone terminal appended behind a live tail: one that refuses even that leaves the run unended (the inline runner claims it again without bound, the queue worker leaves it unclaimed). With two terminals written at once the first is kept and the second is dropped; the caller sees the first one’s outcome. A failed run that is requeued under its own id keeps its earlier run:error in the journal, followed by run:requeued and what the next attempt writes, so a journal can hold a run:error that is not its last event; a retry copies it that way. Only run:requeued and a further run:error may follow a run:error: a store journals a second one when a requeue finds its budget spent, and a later requeue reopens the run.

A store returns headers from load_run, replay_run, read_entries_from and get_last_event (empty mapping when none). Custom stores must too; EventLogStoreContract checks it. The Postgres archive keeps them in a headers jsonb column; an older schema is refused with ArchiveSchemaMismatch.

event_to_dict and event_from_dict in pyrula.contracts.events are the file form that export_run_history writes, always with all six keys. A file with any other key raises JournalNeedsNewerRuntime.

Reserved value keys

An object key starting with __pyrula_ is reserved in every value entering the journal: submit params and metadata, step and ctx.once results, ctx.emit, signal/update/interrupt payloads, resume values, tool inputs and results. At any depth it raises ReservedValueKey (pyrula.contracts.errors) before anything is journaled. HTTP routes answer 422 invalid_request; Cloud commands get the same code. A model’s tool input with the prefix doesn’t fail the run: the tool isn’t called and the model sees the refusal as its error.

Strings containing the text are fine, and so is raw bytes to a Kafka workflow (wrapped in the framework’s own __pyrula_bytes_b64__ marker, which callers can’t forge). A run that ends on the refusal reports invalid_request, not poison. fatal and poison stay the codes for a body that raised Fatal or Poison.

Where control requests land

request_cancel and request_pause append run:cancel / run:pause_requested the moment the store gets them. The worker buffers its own events and writes in batches (event_buffer_size, default 64). So a request that lands mid-step can sit before that step’s step:pending.

At its next boundary the run writes the cancelled run:error or run:paused after everything it did, stamped with the request’s time (cancelled_at, paused_at). Use that event for when it took effect; the request marker is when it was asked for.