HITL basics

In this guide you will add human-in-the-loop (HITL) checkpoints to a workflow using the SDK's operator_confirm() helper. When called, the run suspends completely — the process is idle, no resources are held — and only continues when an operator confirms via the UI or API.

Prerequisites

• A working workflow (see Build your first workflow)
• unitelabs-sdk[automate] installed

When to use HITL

Use operator_confirm() when the workflow cannot safely proceed without a human action:

• An operator needs to physically load a plate before liquid handling begins
• A supervisor needs to visually confirm instrument state before dispensing
• An automatic check fails and a human needs to intervene before retrying

If you only need to notify someone without pausing execution, use logging instead.

Pattern 1 — Unconditional checkpoint

The simplest form: pause at a fixed point and wait for an operator to confirm before continuing. The message you pass is logged with an OPERATOR ACTION REQUIRED: prefix and shown to the operator — make it specific and actionable.

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


@phase(name="Dispense Workflow")
async def dispense_flow() -> dict:
    # ... initialization and setup ...

    # Checkpoint: operator confirms before dispensing begins
    await operator_confirm(
        "Confirm that plate_1 is seated at ByonoyCarrier slot 1 "
        "before dispensing, then click Resume."
    )

    # ... dispensing logic ...
    return {"status": "success"}

When the flow reaches operator_confirm(), the run transitions to AWAITING_INPUT and the platform shows a confirm-and-resume control. The operator resumes it from the platform UI or via the API.

Pattern 2 — Conditional pause on failure

A more powerful pattern: catch a failure, pause for the operator to fix it, then retry the failed step on resume. This is the standard approach for transient physical errors — a plate out of position, a sensor misfiring, a gripper not gripping cleanly.

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


class PlateNotDetectedError(Exception):
    pass


async def detect_plate(carrier_id: str) -> dict:
    """Returns detection result or raises PlateNotDetectedError."""
    ...


@phase(name="Detection with HITL Recovery")
async def detection_flow(carrier_id: str) -> dict:
    try:
        result = await detect_plate(carrier_id=carrier_id)
    except PlateNotDetectedError:
        # Pause — operator physically fixes the plate and confirms; retry on resume
        await operator_confirm(
            f"No plate detected at {carrier_id} slot 1. "
            "Check that the plate is fully seated, then click Resume."
        )
        result = await detect_plate(carrier_id=carrier_id)

    return {"status": "success", "target_slot": result["slot"]}

Chaining multiple checkpoints

For multi-step operator interactions, call operator_confirm() at each checkpoint in sequence:

@phase(name="Multi-Step HITL")
async def multi_step_flow() -> dict:
    await operator_confirm("Step 1/2: Load tip boxes and reservoir. Click Resume when ready.")
    await operator_confirm("Step 2/2: Load sample plate at slot 1. Click Resume to start.")

    # ... instrument logic ...
    return {"status": "success"}

Pattern 3 — Resume via API

An operator does not need to be watching the platform UI. Any system that can call the API can resume a paused run.

operator_confirm() puts the run in AWAITING_INPUT, so resume it by submitting the confirmation (its confirmed field defaults to true) keyed by the waiting phase ID:

curl -X POST "$BASE_URL/v1/runs/{runId}/status" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "status": "RUNNING",
    "data": { "phase-abc-123": { "confirmed": true } }
  }'

Look up the phase ID from the run's status response (the input field). See Typed operator inputs for the full resume-with-data flow. This lets you build operator-facing tools — a lab dashboard, a mobile approval screen, a Slack button — that resume the run when a human clicks confirm.

Handling timeouts

operator_confirm() defaults to a one-hour timeout. Pass timeout= (seconds) to fail the run cleanly if the operator never responds, rather than leaving it paused indefinitely:

await operator_confirm("Load plate at slot 1, then resume.", timeout=600)  # fails after 10 minutes

A timed-out pause transitions the run to FAILED with a clear message in the logs — easier to debug than a run that is silently stuck.

Next steps

• Typed operator inputs: collect structured data — not just a confirmation — with a RunInput model
• Error recovery: clean up hardware state and retry after failure