# Starting and talking to runs

> WorkflowClient: submit runs, signal, query, update, cancel and wait for results.

Source: https://docs.pyrula.io/workflows/guides/client/

```python
from pyrula.workflows import ValkeyStore, WorkflowClient

client = WorkflowClient(ValkeyStore(url="valkey://localhost:6379"))

run = await client.submit("charge", {"amount_cents": 500}, idempotency_key="order-7")
result = await run.result(timeout=60)
```

The client needs only the workers' store. `submit` returns a `RunHandle`; rebuild one with `client.handle(name, run_id)`. Everything is async and follows a `continue_as_new` chain to the live run. A retried `signal_id` or `update_id` stays idempotent across hops for `completed_ttl` (a day by default on Valkey).

| `RunHandle` method | Does | HTTP route |
|---|---|---|
| `signal(signal, payload, *, signal_id)` | Wake `ctx.wait_for_signal`; returns `"delivered"` or `"buffered"` | `POST .../runs/{run_id}/signal` |
| `query(query, *args, **kwargs)` | Read state through `ctx.on_query`; writes nothing | `POST .../query` |
| `update(update, payload, *, update_id, timeout=30.0)` | Apply `ctx.on_update` and return its value | `POST .../update` |
| `resume(action, value, *, wait_id)` | Answer `ctx.interrupt` | `POST .../resume` |
| `cancel(*, reason)` | Request cancellation and queue parked runs for cleanup | `DELETE /{name}/stream/{run_id}` |
| `terminate(*, reason)` | End the run now with no worker; returns `TerminateResult` | `POST .../runs/{run_id}/terminate` |
| `retry()` | New run that reuses a failed run's recorded steps ([Replay and recovery](https://docs.pyrula.io/workflows/concepts/replay-and-recovery.md)) | - |
| `status()` | Current `RunStatus` | `GET .../runs/{run_id}` |
| `result(*, timeout, on_interrupt)` | Wait for the run to finish | - |
| `events()` | The run's events, oldest first | `GET .../runs/{run_id}/events` |

| `WorkflowClient` method | Does | HTTP route |
|---|---|---|
| `submit(name, params, *, idempotency_key, owner, metadata, allow_unregistered=True, start_at, delay)` | Queue a run, optionally for later; returns its handle | `POST /{name}/runs` (takes no start time yet) |
| `trigger(name, event)` | Feed a `Collect`/`Debounce` buffer; returns a handle when a batch fires | `POST /{name}/events` |
| `handle(name, run_id)` | A handle to an existing run | - |

The client also has each handle method with `name, run_id` in front, for one-off calls.

`result(on_interrupt=handler)` calls `handler(payload, run_id)` for each `ctx.interrupt` and resumes with the return value: an `InterruptResponse(action, value)` with `action` one of `approve`, `reject`, `respond` or `edit`, or any other value as a `respond` answer. An unknown action raises `ValueError` and leaves the run suspended. On an `expects="approval"` interrupt, the bare string `"approve"` or `"reject"` is shorthand for that action.

- `submit` with `idempotency_key` returns the first run on a repeat. `owner` tags the tenant for [fair claims](https://docs.pyrula.io/workflows/guides/worker.md#fairness-across-tenants). Version and [flow control](https://docs.pyrula.io/workflows/guides/flow-control.md) are stamped from the decorator. A process that submits by name without importing the module gets no version pin, and takes the `Limit` and `Rate` a serving worker recorded in the store (reread at most every 10 seconds).
- `start_at` and `delay` hold the body back ([Starting a run later](#starting-a-run-later)).
- `update` waits for whichever worker applies it. Past `timeout` it raises `UpdateTimeout`; the update stays queued, so retry with the same `update_id`.
- `resume` answers the interrupt the run is waiting on unless you pass `wait_id`.

## Ids in a long chain

Keep the start id `submit` returned. It reaches the live run while the chain is open and for `completed_ttl` after; an id the chain has moved past works for at least `completed_ttl` after its successor took over, then raises `RunNotFoundError`. `ctx.run_id` is the current run's id, so a body that needs a handle to itself carries the start id in its params.

Before a run is submitted, only `cancel` and `pause` are held. An `update` to such an id raises `RunNotFoundError`; a `signal` is refused.

## Errors

| Exception | When |
|---|---|
| `RunNotFoundError` | No such run, or an update to an id that was only cancelled or paused before it was submitted |
| `TerminateUnsupported` | `terminate` on a store without the `terminate` capability |
| `RunFinishedError` | `signal` or `update` on a finished run; `signal` also once a cancel is requested |
| `RunNotWaitingError` | `resume` when the run isn't waiting on that interrupt |
| `RunNotStartedError` | `update` to a run still waiting for its `start_at` or `delay` |
| `IdempotencyConflict` | `submit` reusing an `idempotency_key` with a different `start_at` or `delay` |
| `QueryError` | Unknown query name |
| `UpdateRejected` / `UpdateError` / `UpdateTimeout` | Validator refused / handler raised / no worker applied it in time |
| `DurableInvocationError` / `DurableInvocationCancelled` / `DurableInvocationTimeout` | `result` on a failed / cancelled / unfinished run |

All are importable from `pyrula.workflows`.

## In the same process

`worker.client` is a client on the worker's store. Its `submit` raises `BackpressureRejected` while the worker's Cloud egress is behind.

```python
import asyncio
from pyrula.workflows import MemoryStore, WorkerOptions, WorkflowWorker, workflow


@workflow(name="cart")
async def cart(ctx) -> list[str]:
    items: list[str] = []

    async def add(sku: str) -> int:
        items.append(sku)
        return len(items)

    ctx.on_update("add", add)
    ctx.on_query("items", lambda: items)
    await ctx.wait_for_signal("checkout")
    return items


async def main() -> None:
    worker = WorkflowWorker(
        workflows=[cart], store=MemoryStore(), options=WorkerOptions(claim_poll_interval=0.05)
    )
    loop = asyncio.create_task(worker.run())
    client = worker.client

    run = await client.submit("cart")
    while await run.status() != "signal_waiting":
        await asyncio.sleep(0.05)
    assert await run.update("add", {"sku": "A-1"}) == 1
    assert await run.query("items") == ["A-1"]
    await run.signal("checkout")
    assert await run.result(timeout=30) == ["A-1"]

    worker.stop()
    await loop


asyncio.run(main())
```

## From sync code

`pyrula.workflows.sync` has a blocking `WorkflowClient` and `RunHandle` for Django and Flask views, Celery tasks and RQ jobs. They are named like the async ones, as with `httpx.Client` and `httpx.AsyncClient`. Workflow bodies stay `async def`; only the caller changes.

```python
import asyncio
import threading

from pyrula.workflows import MemoryStore, WorkerOptions, WorkflowWorker, workflow
from pyrula.workflows.sync import WorkflowClient


@workflow(name="sync_greeting")
async def sync_greeting(ctx, who: str) -> str:
    return await ctx.step("greet", lambda: f"hello {who}")


store = MemoryStore()
worker = WorkflowWorker(
    workflows=[sync_greeting], store=store, options=WorkerOptions(claim_poll_interval=0.05)
)
threading.Thread(target=lambda: asyncio.run(worker.run()), daemon=True).start()

with WorkflowClient(store) as client:
    run = client.submit("sync_greeting", {"who": "ada"})
    assert run.result(timeout=30) == "hello ada"
    rows, _ = client.list_runs("sync_greeting")
    assert [row["run_id"] for row in rows] == [run.run_id]
```

Create one client per process, at import time, and reuse it. It owns one event loop on a daemon thread, started on the first call, and every thread of a threaded server can share it. Errors are the async client's own: `result(timeout=)` raises `DurableInvocationTimeout`.

The methods are the async client's (signatures in the [sync client reference](https://docs.pyrula.io/reference/workflows-sync.md)), plus `list_runs`, which calls the store's own `list_runs` on the calling thread. The async constructor's `admit=` hook is not offered.

- A call cut short may still have taken effect. Ctrl+C while waiting re-raises `KeyboardInterrupt`, and `close()` from another thread makes a waiting call raise `RuntimeError`. After either, retry with the idempotency parameters: `idempotency_key` on `submit`, `signal_id` on `signal`, `update_id` on `update`.
- `result(on_interrupt=)` handlers run on the client's loop thread, shared by every caller. A handler that blocks stalls them all.
- `close()` stops the thread and is safe to call twice; `with` calls it for you. A call after it raises `RuntimeError("client is closed")`. A client left open is closed at interpreter exit.
- A sync method called from a thread running an event loop raises `RuntimeError`. Use `pyrula.workflows.WorkflowClient` there.
- A client created before `os.fork()` (Celery prefork, gunicorn with a preloaded app) builds its own loop and thread in the child. Fork while no thread is inside a store call, and open stores whose sockets don't survive a fork (Valkey) after it. `SQLiteStore` reopens its connections per process, but a fork taken mid-call has failed in the child with `disk I/O error`.

## Starting a run later

`delay` (seconds or `timedelta`) or `start_at` (timezone-aware `datetime`) holds the body until then. One of them, not both.

```python
import asyncio
from datetime import timedelta

from pyrula.workflows import workflow
from pyrula.workflows.testing import WorkflowTestEnv


@workflow(name="send_receipt")
async def send_receipt(ctx, order: str) -> str:
    return f"receipt for {order}"


async def main() -> None:
    env = WorkflowTestEnv()
    run = await env.client.submit("send_receipt", {"order": "o-1"}, delay=timedelta(hours=2))
    assert await run.result() == "receipt for o-1"   # the two hours are virtual


asyncio.run(main())
```

```python
from datetime import UTC, datetime, timedelta

await client.submit("send_receipt", {"order": "o-1"},
                    start_at=datetime.now(UTC) + timedelta(days=30))
```

The wait is the run's first sleep: no worker or [concurrency](https://docs.pyrula.io/workflows/guides/flow-control.md) slot held, and it survives restarts. One worker claims the run briefly to arm it. It reads `pending` until armed, then `sleeping`. A `delay` becomes an absolute time on the client's clock at `submit`.

- **Bounds.** A negative, NaN, infinite or boolean `delay`, a naive `start_at`, or a time more than 365 days out raises `ValueError`. A past `start_at` or zero `delay` starts now.
- **Lateness.** The body can start late by `timer_poll_interval_s` plus the claim.
- **Idempotency.** Same `idempotency_key` and same timing returns the first run; different timing raises `IdempotencyConflict`. The key lives as long as the delay.
- **Before it starts.** `cancel` and `terminate` end the run without running the body, and a pause keeps it from beginning. Signals are buffered. An `update` raises `RunNotStartedError` (409 over HTTP).
- **Retries and successors.** A retry does not inherit the delay, and a `continue_as_new` successor starts at once. Schedules and Kafka records take no start time.
- **Deploys.** A newer worker can adopt a waiting run if the run follows versions (`auto_upgrade`, the default when a version is set); a pinned run waits for its version.

## Without a worker

`InlineWorkflowRunner` and `WorkflowTestEnv` have a `client` whose handles drive the run in your process. `WorkflowTestEnv` (from `pyrula.workflows.testing`) keeps a virtual clock and skips to a sleep's wake time and to the deadline of a `wait_for_signal` or `interrupt` with `timeout=`. A wait with no deadline stays parked until the test answers it.

```python
import asyncio
from datetime import timedelta

from pyrula.workflows import workflow
from pyrula.workflows.testing import WorkflowTestEnv


@workflow(name="trial")
async def trial(ctx, user: str) -> str:
    await ctx.sleep(timedelta(days=30))
    return f"{user}: trial over"


async def main() -> None:
    env = WorkflowTestEnv()
    run = await env.start(trial, {"user": "ada"})
    assert await run.result() == "ada: trial over"   # the 30 days are virtual


asyncio.run(main())
```

`await env.advance(timedelta(...))` moves the clock by hand; `env.run_until_complete(run)` drives a run until it ends or waits on a signal.

A child started with `ctx.run_child` runs in the test env while its parent waits on it, and `run_child(timeout=)` counts virtual time. `env.children(run)` returns the handles of the children a run waits on. Signal a parked child through its handle from another task; if every child is parked on input and nothing changes for two seconds, the wait raises a `RuntimeError` naming each child and its signal.

## Cancellation

Cancel is graceful and needs a worker; to end a run now without one, see [Cancel and terminate](https://docs.pyrula.io/workflows/guides/cancel-and-terminate.md).

`await client.cancel(name, run_id)` records a request. A sleeping, signal-waiting, interrupted or paused run is queued for its cancelled terminal with no further work. A running step finishes, and the next live effect reads the store before starting. Recorded results still replay.

Once cancel is requested, `signal` raises `RunFinishedError` (409 over HTTP). Cleanup goes through normal version, tenant, concurrency and rate admission, so a saturated queue or missing worker delays it; a parked run can show `in_progress` until `run:error` with `code="cancelled"` arrives.

Inside a body, `ctx.cancelled` is a hint refreshed by heartbeats (10 s default) and can be stale during outages. Open child runs follow `ctx.run_child`'s `parent_close` (cancelled by default; see [Parallel steps and child runs](https://docs.pyrula.io/workflows/guides/parallel-and-children.md#what-happens-to-a-child-when-its-parent-ends)). A cancelled child surfaces to its parent as `ChildWorkflowError`.