Signals, queries and updates

Pyrula provides three operations for interacting with a running workflow:

	Direction	Changes the run	Returns
Signal	in	yes, when the run reads it	nothing
Query	out	no (read-only replay)	the handler’s value
Update	in, then out	yes	the handler’s value

Signals

ctx.wait_for_signal(name) suspends a run until something sends a signal by that name. The worker is freed while it waits.

from pyrula.workflows import workflow, Signal

@workflow(name="payment_processor")
async def payment_processor(ctx, order_id: str, amount: float) -> str:
    await ctx.step(
        "initiate_payment",
        lambda: initiate_payment(
            order_id,
            amount,
            idempotency_key=f"{ctx.run_id}:initiate_payment",
        ),
    )

    signal: Signal = await ctx.wait_for_signal("payment_confirmed")

    if signal.payload.get("status") == "success":
        await ctx.step(
            "fulfill_order",
            lambda: fulfill_order(order_id, idempotency_key=f"{ctx.run_id}:fulfill_order"),
        )
        return "fulfilled"
    await ctx.step(
        "cancel_order",
        lambda: cancel_order(order_id, idempotency_key=f"{ctx.run_id}:cancel_order"),
    )
    return f"failed: {signal.payload.get('reason')}"

Signal lifecycle

1. The run hits wait_for_signal. The engine emits run:signal_waiting, sets status to signal_waiting, and frees the worker.
2. Something delivers the signal by name with a payload.
3. The engine emits run:signal_received and re-queues the run.
4. On replay the wait point returns the recorded signal and execution continues.

A signal that arrives before the run reaches the wait point is buffered, so there’s no race between the sender and the waiter.

Signal

@dataclass(frozen=True)
class Signal:
    name: str
    payload: dict[str, Any]
    sent_at: str   # ISO-8601 timestamp set by the sender

Send one from another workflow with await ctx.send_signal(target_name, target_run_id, signal_name, payload), over HTTP with POST /{name}/runs/{run_id}/signal (Serving over HTTP), or directly on the store with store.send_signal(name, run_id, signal_name, payload, sent_at).

Queries and updates

A query reads a run’s state without changing it. An update changes it and returns a value, after an optional validator has accepted it.

from pyrula.workflows import workflow

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

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

    def validate(sku: str = "") -> None:
        if not sku:
            raise ValueError("sku required")

    ctx.on_update("add_item", add, validator=validate)
    await ctx.wait_for_signal("checkout")
    return items

Calling them with a WorkflowClient:

run = client.handle("cart", run_id)
count = await run.update("add_item", {"sku": "A-1"})   # 1
items = await run.query("items")                       # ['A-1']

• The payload arrives as keyword arguments.
• Query handlers are sync and read-only. A query replays the run and writes no events, and raises QueryError for an unknown query.
• Update handlers are async. They may mutate the body’s locals and await ctx.step(...), but must not suspend (no sleep, interrupt or wait_for_signal). They run at the run’s next suspension point.
• Validators are sync and read-only. Raising rejects the update before it is recorded; the caller gets UpdateRejected (from pyrula.workflows). A handler that raises gives UpdateError.
• update waits for a worker to apply it, up to timeout (30 s), then raises UpdateTimeout; the update stays queued. Pass update_id to make a retry idempotent.

Over HTTP: POST /{name}/runs/{run_id}/update (a timeout answers 504) and /query.