Workflows

A workflow ties labware threads together. A thread describes the route of one piece of labware; the workflow declares which threads exist, which one starts the run, what variables it accepts, and which event handlers fire. A workflow is what you submit to the runtime.

Your first workflow

import orca.orca as orca
from orca.workflow_models.workflow_context import WorkflowContext

@orca.workflow(name="my_assay")
def my_assay(wf: WorkflowContext):
    wf.start(sample_thread)
    wf.thread(tips_thread)

Two threads. The sample plate starts the run. The tip rack is registered, and spawns when a step needs tips.

Builder call	Declares
wf.start(thread)	An entry thread; spawns when the workflow starts.
wf.thread(thread)	A thread registered for auto-spawn, keyed by the labware template it carries.
wf.variable(name, definition)	A workflow-scoped variable that submissions can override.
wf.on(event_name, handler)	An event handler subscribed for this workflow's executions. Under Advanced below.

The workflow function is a builder. It receives a WorkflowContext (wf) and calls methods on it to declare structure. The function runs once, at decoration time, and the decorator returns a WorkflowTemplate. wf.template returns the WorkflowTemplate being built, if you need it inside the builder.

The builder must be a regular function

@orca.workflow calls your function immediately and raises TypeError if it returns a coroutine. Declare it with def, not async def. There is nothing to await: every call on wf is synchronous registration.

The decorator collects the methods and threads above it

@orca.workflow attaches every @orca.method and @orca.thread decorated since the previous @orca.workflow (or since module import). That is why the common layout is a build_workflow(topology) function that defines actions, methods and threads, then ends with the @orca.workflow builder. Two builder calls each get their own scope, so a method defined for workflow A cannot leak into workflow B.

Method and thread names are unique per workflow, not per deployment. Two workflows may each declare a seal_step.

Entry threads and auto-spawned threads

wf.start(...) declares an entry point: a thread that always spawns when an execution begins. A workflow may have more than one.

wf.thread(...) registers a thread for auto-spawn, keyed by the labware template it carries. When an owning thread resolves an action whose inputs name that template and no live thread is carrying one, the engine spawns this thread to supply it. Two threads registered for the same labware template raise ValueError at registration.

Variables

Workflows declare variables that submissions and operators override:

from orca.variables.variable_definition import VariableDefinition

@orca.workflow(name="dilution_assay")
def dilution_assay(wf: WorkflowContext):
    wf.variable(
        "dilution_factor",
        VariableDefinition(
            type="float",
            default=10.0,
            min=2.0,
            max=100.0,
            description="Serial dilution factor",
        ),
    )
    wf.start(plate_thread)

Inside an action: factor = await ctx.param("dilution_factor", float). ctx.param is async, so await it. Pass the expected type as the second argument for a narrowed return; omit it for the untyped form.

Names starting with global., and the bare name global, are reserved and rejected at registration.

See Variables for the type taxonomy, validation rules, the two scopes, and the resolution order.

Advanced

Everything below is for workflows where several threads feed one piece of labware, where a run reacts to events, or where the same receiver is used more than once.

Threads that cannot be an entry point

Two thread shapes are refused as wf.start(...) entries at build time: a thread whose start uses REUSE_EXISTING, and a thread declared immovable=True. Both describe labware that is already on the deck and stays there, which is a contributor, never a starting point. See Labware threads.

Capacity and overflow

wf.thread(thread, capacity=None, overflow_strategy=None) also accepts a capacity policy for the thread's receiver slot. overflow_strategy attaches a custom IOverflowStrategy.

from orca.resource_models.capacity import CapacityPolicy, OverflowAction

wf.thread(
    final_plate_thread,
    capacity=CapacityPolicy(max_contributions=4, overflow_action=OverflowAction.NEW),
)

max_contributions is an upper bound, not a guarantee. The engine reuses the active receiver only while the labware still reports can_continue() and the slot still has room, so a rack that runs out of tips overflows early. OverflowAction is NEW (spawn a fresh receiver), REJECT (fail the workflow) or RECOVERABLE_REJECT (raise to an operator decision).

A receiver whose labware runs out ends there and hands off, even with slot room to spare, and the contribution it could not serve reaches its successor. See Labware threads for what the successor adopts and what it asks the operator for.

Joining threads

A contributor thread yields orca.join(...) to say it participates in another thread's method instead of running one of its own:

@orca.thread(labware=tips, start="stacker", end="waste")
async def tips_thread(ctx: ThreadContext):
    yield orca.join(allows=[transfer_step, dilute_step])

The thread waits where it is until a method that lists its labware dispatches to it, then joins that method as a contributor. allows=[...] names the method templates it will accept; joining anything else raises ValueError. orca.join(some_method) pins one specific method, and a bare orca.join() accepts whatever spawned the thread.

See Methods for what the owner and the contributors each do once they meet.

Waiting on an event

orca.on(event_name, timeout=None) is a step a thread yields. It blocks until the named event has been published:

@orca.thread(labware=plate, start="stacker", end="waste")
async def assay(ctx: ThreadContext):
    yield prepare_step
    yield orca.on("reagent_ready")
    yield run_step

orca.on reads from counter zero every time, so it sees the latest publish whether or not the thread was already waiting when the event fired. timeout=None waits indefinitely; a timeout that expires pauses the thread for operator recovery.

Branching on events

orca.branch(...) picks one of several method sequences from an emitted event value:

yield orca.branch("qc_result", {
    "pass": [seal_step],
    "fail": [retry_step, evaluate_qc_step],
    "else": [abort_step],
}, timeout=300.0)

The branch waits for the named event, converts the published value to a string, and runs the matching list. The "else" key is the fallback. A value with no matching branch and no "else" raises ValueError.

Event handlers

from orca.events.event_handler_interface import IEventHandler
from orca.events.execution_context import ExecutionContext

class MyAuditHandler(IEventHandler):
    def handle(self, event: str, context: ExecutionContext) -> None:
        ...

@orca.workflow(name="my_assay")
def my_assay(wf: WorkflowContext):
    wf.start(plate_thread)
    wf.on("THREAD.CREATED", MyAuditHandler())

A handler is either an IEventHandler (one handle(event, context) method) or a plain callable with the same two arguments. It is subscribed to the runtime event bus under that event name when an execution of this workflow starts, and is called synchronously on every matching event.

ctx.emit("qc_result", ...) inside an action, method or thread also reaches the bus, as CUSTOM.qc_result.EMITTED. See Events for the full name taxonomy and the context payloads.

See also

• Labware threads: defining the route one piece of labware takes.
• Variables: the types a workflow variable can have and how one resolves.
• Submissions: submitting a workflow with groups and variables.
• Runtime: the process that holds your workflows and runs them.
• Events: the full event taxonomy a handler can subscribe to.