Human in the Loop

Human in the loop (HITL) is a pattern for semi-automated workflows: the workflow engine pauses a run at a defined point, waits for an operator to confirm an action or provide input, then continues.

The SDK provides two levels of API:

• operator_confirm() — a one-line helper for the common case: the operator just needs to confirm a physical action was taken.
• pause_flow_run() — the lower-level pause function, used directly when the operator has to enter structured data and not just confirm.

Simple confirmation

The most common case: pause and wait for an operator to confirm a physical action — no structured data needed. Use operator_confirm():

from unitelabs.sdk import operator_confirm
from unitelabs.sdk.automate import phase

@phase()
async def load_plate():
    await operator_confirm("Load sample plate into slot 1, then click Resume.")

operator_confirm() logs the message with an OPERATOR ACTION REQUIRED: prefix (easy to spot in the run log), pauses the run until the operator confirms, then logs that the run resumed. Pass timeout= (seconds, default 3600) to fail the run cleanly if no one responds.

It requires the automate extra (unitelabs-sdk[automate]); without it the call raises RuntimeError. Internally it pauses with a single-field confirmation model (OperatorConfirmation), which is what puts the run into AWAITING_INPUT instead of a bare PAUSED state.

Typed input

When the operator needs to provide structured data, pass a RunInput subclass to pause_flow_run().

RunInput is a base class the SDK exports (from unitelabs.sdk import RunInput). Subclass it and declare one field per value you want back. Since it is a pydantic model, the SDK sends its JSON Schema along with the pause: the platform picks a form control per field from that schema, and on submit pydantic validates the values and hands you back an instance of your class.

from unitelabs.sdk import pause_flow_run, RunInput
from unitelabs.sdk.automate import phase

class QCInput(RunInput):
    sample_id: str
    approved: bool
    notes: str = ""

@phase()
async def quality_check():
    data = await pause_flow_run(wait_for_input=QCInput)
    if not data.approved:
        raise ValueError(f"Sample {data.sample_id} rejected at QC gate. Notes: {data.notes}")
    await log_approval(data.sample_id)

The run transitions to AWAITING_INPUT. The operator opens the run in the platform, fills in the form, and submits. Execution continues with the submitted values available as data.

Instructions and pre-filled fields

operator_confirm() takes a message. pause_flow_run(wait_for_input=QCInput) takes none, so on its own it leaves the operator with a form and no task. Pass the instruction through with_initial_data():

data = await pause_flow_run(
    wait_for_input=QCInput.with_initial_data(
        description="Inspect the plate under the scope, then submit.",
        sample_id=current_sample,
    )
)

description accepts Markdown and travels with the input request, next to the schema in the run status. Every other keyword argument pre-fills that field, so the operator only edits what is still open.

The Python type annotation on each field determines which form control the platform renders:

Fields without a default value are required. Fields with a default value are optional and pre-filled.

For constrained choices (e.g., selecting from a fixed list of protocols), use Literal or a Python Enum — both render as a dropdown. See Typed operator inputs for detailed examples.

Run state during a pause

AWAITING_INPUT means the platform is holding a form open. operator_confirm() uses a single pre-filled confirmation field, so the form is effectively a confirm-and-resume button; a custom RunInput model renders one control per field. Bare pause_flow_run() carries no schema and the operator simply clicks Resume.

Related concepts

• Input: how workflow parameters become operator-facing fields
• Runs: the AWAITING_INPUT and PAUSED run states and resumption

Guides

• Basic human in the loop
• Typed operator inputs