Error Handling

Error handling in UniteLabs is built into the workflow hierarchy. Different error types are handled at the level best suited to recover from them — automatic retries at the step level, and try/except with operator-assisted recovery at the phase level.

Error types

Step level: automatic retry

Steps are the first line of defense. The workflow engine retries technical errors automatically before surfacing them:

from unitelabs.sdk.automate import step

@step(retries=3, retry_delay_seconds=2)
async def aspirate(liquid_handler: LiquidHandlerDevice, volume: float):
    await liquid_handler.pipettes.aspirate(volume)

If the step fails after all retries, the error is propagated to the enclosing phase.

The default retry count is configurable per step. A retry re-runs the step from the top, so a step that already did half its work does that half twice: retry a dispense that filled three wells before failing, and those three wells get filled again. Which steps are safe to retry is a design decision you make when you write them, and Error recovery walks through it.

Phase level: try/except and operator recovery

At the phase level, use try/except to catch errors and decide how to respond — retry, pause for operator intervention, or abort cleanly:

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


class PlateNotDetectedError(Exception):
    pass


@phase()
async def detection_phase(lh, carrier_id: str) -> dict:
    logger = get_logger()
    try:
        result = await detect_plate(lh=lh, carrier_id=carrier_id)
    except PlateNotDetectedError:
        logger.warning(
            f"No plate detected at {carrier_id}. "
            "Seat the plate fully, then click Resume."
        )
        await pause_flow_run()
        logger.info("Operator resumed — retrying detection.")
        result = await detect_plate(lh=lh, carrier_id=carrier_id)
    return {"lh": lh, "target_slot": result["slot"]}

Use try/finally to guarantee cleanup regardless of outcome:

@phase()
async def dispensing_phase(lh) -> dict:
    try:
        lh = await distribute(lh=lh, volume=100.0)
    finally:
        lh = await return_tips(lh=lh)  # always runs — success, error, or cancellation
    return {"lh": lh}

See Error recovery for more patterns.

Workflow level: propagation and setting a checkpoint

If a phase fails and no try/except handles the error, the failure propagates to the workflow. The run is marked as FAILED and the workflow engine records the last completed phase as a checkpoint.

Because phases always end in stable states, a failed run can be resumed from the last successful phase — either automatically or by an operator.

Summary

Step fails
  └── Technical error → retry (up to N times)
        └── Still failing → propagate to phase

Phase receives error
  └── try/except → pause for operator, retry, or return error
  └── No handler → propagate to workflow

Workflow receives error
  └── Run marked FAILED at last stable checkpoint
  └── Operator can resume from checkpoint

Related concepts

• Step: where retries are configured
• Human in the Loop: using pause_flow_run() for operator-assisted recovery
• Runs: run states and checkpoint-based resumption

Guides

• Basic error handling
• Advanced error handling