Guides

Error recovery

Clean up hardware state, retry transient failures, and resume after operator intervention.

In lab automation, an error mid-run can leave hardware in an inconsistent state — tips are still loaded, a plate is mid-transfer, a gripper is holding something. Error recovery means: clean up first, then decide whether to retry or abort.

This guide covers four practical patterns. They compose well — most real workflows use two or three of them together.

For unitelabs-sdk< 0.10.0, import flow and task from Prefect instead. Use flow in place of both workflow and phase decorators, and task as step decorators.

Prerequisites

A working workflow (see Build your first workflow)

Pattern 1 — Guaranteed cleanup with `try/finally`

Use try/finally to ensure hardware is left in a safe state even when an operation fails. The finally block always executes — whether the try body succeeds, raises, or is cancelled.

from unitelabs.sdk import get_logger
from unitelabs.sdk.automate import phase


@phase(log_prints=True, name="Dispensing Phase")
async def dispensing_phase(lh) -> dict:
    logger = get_logger()
    try:
        lh = await distribute_colour(lh=lh, reservoir="ice", volume=100.0)
        lh = await distribute_colour(lh=lh, reservoir="orange", volume=100.0)
    finally:
        # Always return tips to the rack — even if distribute raises.
        # This runs on success, exception, and Prefect cancellation.
        lh = await return_tips(lh=lh)
        logger.info("Tips returned.")

    return {"lh": lh, "status": "success"}

Use finally for any action that must happen regardless of outcome: returning tips, closing connections, unlocking an instrument, setting a status flag. Do not use it for logic that should only run on success — put that in the try body.

Pattern 2 — Automatic retry with delay

Add retries and retry_delay_seconds to a @phase or @step to have the SDK automatically re-run it when an unhandled exception escapes. This is ideal for transient hardware failures — a gripper that occasionally mis-grips, a sensor that times out under load.

from unitelabs.sdk.automate import phase


@phase(
    log_prints=True,
    name="Return Plate Phase",
    retries=1,
    retry_delay_seconds=60,
)
async def return_plate_phase(lh) -> dict:
    """
    Move plate from working carrier back to storage.
    If the gripper motion fails, Prefect retries once after 60 seconds.
    The delay gives the instrument time to reset before the next attempt.
    """
    lh = await gripper_move(
        lh=lh,
        source="working_carrier[1]",
        destination="storage_carrier[4]",
    )
    return {"lh": lh}

The same decorator works on @step:

from unitelabs.sdk.automate import step


@step(
    log_prints=True,
    name="Fetch Instrument Data",
    retries=3,
    retry_delay_seconds=10,
)
async def fetch_instrument_data_task(endpoint: str) -> dict:
    # Retried up to 3 times on any exception, with 10s between attempts
    ...

Only retry idempotent operations. Retrying a dispense that partially succeeded will double-dispense already-filled wells. Retrying a plate move when the plate was already moved will crash. When in doubt, use Pattern 3 instead: pause and let a human decide.

Pattern 3 — Pause-and-retry (HITL recovery)

When a failure requires human intervention before retrying — not just waiting — combine try/except with pause_flow_run(). The operator fixes the physical issue, clicks Resume, and the flow retries from that point.

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


class PlateNotDetectedError(Exception):
    pass


@phase(log_prints=True, name="Detection Phase with Recovery")
async def detection_phase(lh, carrier_id: str) -> dict:
    logger = get_logger()

    try:
        result = await detect_plate(lh=lh, carrier_id=carrier_id)
        logger.info(f"Plate detected at slot {result['slot']}.")
    except PlateNotDetectedError:
        # 1. Log a clear, operator-readable message before pausing
        logger.warning(
            f"No plate detected at {carrier_id} slot 1. "
            "Check that the plate is fully seated, then click Resume."
        )
        # 2. Suspend the run — no resources held while waiting
        await pause_flow_run()
        # 3. On resume, retry detection (plate is now in position)
        logger.info("Operator resumed — retrying detection.")
        result = await detect_plate(lh=lh, carrier_id=carrier_id)

    return {"lh": lh, "target_slot": result["slot"]}

The run stays PAUSED until the operator acts. They can resume from the Prefect UI or via the API. See HITL basics for more on the pause mechanism.

Pattern 4 — Phase tracking for partial recovery

Track which phases completed successfully in a list. Include it in every return value — success and error alike. This tells you exactly where a run failed, which phases are safe to skip on a re-run, and what to report to a LIMS.

from unitelabs.sdk import get_logger
from unitelabs.sdk.automate import workflow


@workflow(log_prints=True, name="Full Workflow")
async def full_workflow(lh) -> dict:
    logger = get_logger()
    phases_completed: list[str] = []

    # Phase 01
    lh = await initialization_phase(lh=lh)
    phases_completed.append("01_initialization")

    # Phase 02
    result = await detection_phase(lh=lh)
    lh = result["lh"]
    phases_completed.append("02_detection")

    # Phase 03 — dispensing can fail mid-run
    try:
        lh = await dispensing_phase(lh=lh)
        phases_completed.append("03_dispensing")
    except Exception as e:
        logger.error(f"Dispensing failed: {e}")
        return {
            "status": "error",
            "failed_phase": "03_dispensing",
            "phases_completed": phases_completed,
            "message": str(e),
        }

    phases_completed.append("04_finalization")
    return {"status": "success", "phases_completed": phases_completed}

Knowing phases_completed = ["01_initialization", "02_detection"] means Phase 03 was the one that failed, Phases 01 and 02 do not need to be repeated, and the instrument state after Phase 02 is the starting point for a manual recovery.

Graceful vs fail-fast

For flows triggered by external systems (a UI, a webhook, a scheduler), the response format matters:

Pattern	Prefect run status	Use when
`try/except` → return error dict	`Completed`	Caller expects a structured response
Let exception propagate	`Failed`	You want Prefect alerts and on-call paging

# Graceful — caller sees {"status": "error", ...} instead of an exception
@workflow(log_prints=True, name="Graceful Flow")
async def graceful_flow(source: str) -> dict:
    try:
        result = await process(source=source)
        return {"status": "success", **result}
    except ValueError as e:
        return {"status": "error", "message": str(e)}

For a deeper treatment of structured error types and error codes, see Advanced error handling.

Next steps

HITL basics: pause_flow_run() in depth
Advanced error handling: typed error taxonomy with user-facing messages and error codes

Typed operator inputs

Collect structured data from operators at runtime using a RunInput model — enums, booleans, numbers, and text fields rendered as a platform form.

Advanced error handling

Build a structured error taxonomy with typed error codes, user-facing messages, and automatic exception classification.