Signals, queries and updates

	Direction	Changes the run	Returns
Signal	in	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 a signal by that name arrives, freeing the worker.

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')}"

The run emits run:signal_waiting and frees the worker. Delivery emits run:signal_received and re-queues; replay returns the recorded signal at the wait. A signal that arrives before the run reaches the wait is buffered.

Waiting with a timeout

timeout= (seconds or timedelta) or deadline= (timezone-aware datetime), not both. Negative timeout or naive deadline: ValueError.

from pyrula.workflows import WaitTimeout

try:
    signal = await ctx.wait_for_signal("payment_confirmed", timeout=3600)
except WaitTimeout:
    return "payment never confirmed"

At the deadline wait_for_signal raises WaitTimeout (wait_kind, name, call_index, deadline). Uncaught, the run ends in error with code wait_timeout.

• A signal already in the store before the timer ran wins, even past the deadline (bounded by the timer’s poll interval).
• A signal sent after the timeout is buffered for the next wait_for_signal of that name.
• The deadline is fixed when the wait opens and recorded in run:signal_waiting; restarts and replays reuse it.
• timeout=0 or a past deadline takes a buffered signal if there is one, else times out at once.
• No waiting inside a ctx.gather leg, see Parallel steps.

Sending

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

status = await client.signal("payment_processor", run_id, "payment_confirmed", {"status": "success"})
# "delivered" or "buffered"; signal_id= makes a retried send idempotent

A signal to a finished run raises RunFinishedError, as does one after a cancel was requested. From another workflow: await ctx.send_signal(target_name, target_run_id, signal_name, payload). Over HTTP: POST /{name}/runs/{run_id}/signal (Serving over HTTP).

Queries and updates

A query reads run state without changing it. An update changes it and returns a value, after an optional validator accepts 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

Call 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.
• Queries are sync and read-only. A query replays the run, writes no events, works on finished runs, and raises QueryError for an unknown name.
• Update handlers are async. They can mutate the body’s locals and await ctx.step(...) but must not suspend. They run only at a suspension point (sleep, wait_for_signal, interrupt); a body that never reaches one while an update waits refuses it.
• Validators are sync and read-only. Raising rejects the update before it’s recorded (UpdateRejected). A handler that raises gives UpdateError.
• update waits up to timeout (30 s) for a worker, then raises UpdateTimeout; the update stays queued. update_id makes a retry idempotent.
• An update a run can’t apply stays buffered, never dropped. A paused run applies it on resume; a run with no handler never does, and the caller gets UpdateTimeout.
• A worker that dies holding a run with a buffered update doesn’t wake it at lock expiry; the orphan sweep (60 s default) recovers it.
• A successful update ran before the body returned. One still waiting when the body returns, or sent to a finished run, gets RunFinishedError.

Over HTTP: POST /{name}/runs/{run_id}/update (timeout answers 504) and /query, bodies in Serving over HTTP.