# Runs as Event Streams

> One run is one append-only event stream, used both as the replay log and the client feed.

Source: https://docs.pyrula.io/workflows/concepts/runs-as-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`.

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

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