Tasks and Transport

Overview

This section covers tasks, the fundamental building blocks of workflows. It explains the various types of tasks and their relationship to transport capabilities within a workflow.

Tasks are any action that must occur in a sequence in a workflow. A LINQ workflow is, fundamentally, a graph of tasks that take place on instruments with labware or execute via custom code or data integrations.

ActionTask basics

The ActionTask class represents the execution of an action on a workcell. This could be the positioning of labware from one point to another, an instrument performing an operation via a driver, or a step requiring a manual operator confirmation.

uncap_tubes = ActionTask(
    id="uncap_tubes",
    description="Uncap tubes",
    instrument_type=uncapper.type,
    action="uncap",
    time_estimate=30,
    labware_sources=[centrifuge_tubes.forward(labware=tubes)],
    labware_outputs=[LabwareOutput(tubes)],
)

Both ActionTask and CodeTask support the optional available_failure_actions parameter. See Error Handling for configuration details and runtime semantics.

For complete API reference and all ActionTask properties, see Tasks.

For the relationship of ActionTasks to instruments and drivers, see Instruments and Drivers.

For controlling the order and timing of tasks, see Time and Transport.

For tasks involving a technician mid-workflow, see Operator Intervention.

For mid-workflow conditional task execution, see Workflow Branching.

Data Connector

In the SDK, you can export data through tasks that connect to external services such as SFTP, HTML, S3, LIMS, ELN, and more. Tasks can link to data connectors, which facilitate the transfer of data to and from external systems.

For more details on how data connectors work, see the Data Manipulation section.

CodeTask

CodeTask enables execution of custom logic within a workflow.

For tasks involving custom code execution, see Data Manipulation. For failure behaviour configuration, see Error Handling.

Implicit Actions

Some tasks may implement multiple driver actions for a given instrument that frequently occur in sequence. A typical example is loading: devices may require sending an open_door command, moving labware into the loading bay, sending a close_door command, and then issuing a mount command.

While these tasks can be modeled individually, certain high-level actions (such as load) are compiled into the required sequence of driver actions automatically. This behavior is configured per instrument and per driver.

Implicit Transport

When defining a workflow, you only need to specify instrument tasks (e.g. seal → centrifuge → peel). Transport tasks — movements performed by the SCARA arm and LINQ transport layer — are derived automatically from labware inputs and outputs.

For more detail on controlling the movement of labware through instrument slots, see Labware Flow.

For detail about estimating transport timing for planning and simulation, see Time and Sequence.

Controlling Labware Transport Order

In scenarios where labware must be moved in a specific sequence (for example, ensuring one plate is placed on an instrument before another), dependencies can be set on labware sources.

load_plate_liquid_handler = ActionTask(
    id="liquid_handle_load_2",
    description="load a plate on the liquid handler",
    instrument_type=liquid_handler.type,
    action="do_nothing",
    time_estimate=5,
    labware_sources=[centrifuge.forward(labware=plate_2, dependencies=["liquid_handle_load_1"])],
    labware_outputs=[LabwareOutput(plate_2)],
)

In this example, plate_2 will not be transported to the liquid handler until liquid_handle_load_1 has completed, ensuring correct loading order.

Task Dependencies

Tasks form a directed graph in which a task can depend on one or more other tasks. A task cannot execute until all its dependencies have completed.

Expressing Task Dependencies

Task-level dependencies are expressed using the dependencies parameter on ActionTask and CodeTask. Dependencies can reference:

ActionTask — Instrument actions
CodeTask — Custom code execution
Conditional — Branching logic with outputs

# Task B depends on Task A
task_a = ActionTask(
    id="task_a",
    # ... configuration ...
)

task_b = ActionTask(
    id="task_b",
    # ... configuration ...
    dependencies=[task_a],  # task_b waits for task_a to complete
)

# Task C depends on both Task A and Task B
task_c = ActionTask(
    id="task_c",
    # ... configuration ...
    dependencies=[task_a, task_b],  # task_c waits for both to complete
)

Task Dependency vs. Labware Dependency

Task-level dependencies control when a task can execute. Labware-level dependencies (set via LabwareSource(..., dependencies=[...])) control the order in which specific labware is transported.

These are independent:

Use task dependencies for execution ordering (e.g., “complete sealing before centrifugation”)
Use labware dependencies for transport sequencing (e.g., “load plate_1 before plate_2”)

Complex Dependency Graphs

You can express sophisticated dependency patterns:

# Parallel execution followed by sequential consolidation
analysis_1 = ActionTask(id="analysis_1", dependencies=[prep])
analysis_2 = ActionTask(id="analysis_2", dependencies=[prep])
analysis_3 = ActionTask(id="analysis_3", dependencies=[prep])

consolidate = ActionTask(
    id="consolidate",
    dependencies=[analysis_1, analysis_2, analysis_3]  # Waits for all 3
)

Continuous Loading

Continuous loading enables operators to dynamically load labware during workflow execution, with each loaded item triggering a predefined sequence of tasks. This is useful for processing multiple batches of samples when the exact number of items is not known at workflow creation time.

Basic Continuous Loading

A ContinuousLoading task defines:

The type of labware being continuously loaded
Additional data inputs collected from the operator for each item
A sequence of tasks to execute for each loaded item

from linq.task import ContinuousLoading, AdditionalDataInputString, AdditionalDataInputInteger

continuous_loading = ContinuousLoading(
    id="sample_processing",
    description="Continuous sample processing",
    loaded_labware=sample_plates,
    additional_data_inputs=[
        AdditionalDataInputString(
            id="sample_type",
            label="Sample Type",
            options=["serum", "plasma", "whole_blood"],
            default_value="serum"
        ),
        AdditionalDataInputInteger(
            id="sample_count",
            label="Number of Samples",
            default_value=96
        )
    ]
)

Adding Tasks to Continuous Loading

Add the continuous loading task and subsequent tasks to the workflow as usual:

continuous_loading = ContinuousLoading(
    id="sample_processing",
    description="Process samples as they arrive",
    loaded_labware=sample_plates,
    additional_data_inputs={...}
)

prep_task = ActionTask(
    id="prep_samples",
    description="Prepare samples for analysis",
    instrument_type=prep_station.type,
    action="prepare",
    time_estimate=120,
    labware_sources=[StoredLabware(sample_plates, slot=1)],
    labware_outputs=[LabwareOutput(sample_plates)]
)

analyze_task = ActionTask(
    id="analyze_samples",
    description="Analyze prepared samples",
    instrument_type=analyzer.type,
    action="analyze",
    time_estimate=300,
    labware_sources=[prep_task.forward(labware=sample_plates)],
    labware_outputs=[LabwareOutput(sample_plates)]
)

workflow = Workflow(
    tasks=[continuous_loading, prep_task, analyze_task, ...]
)

Important: Only one continuous loading task is supported per workflow.

Additional Data Input Types

Continuous loading supports collecting additional information from operators for each loaded item.

String Input with Options

AdditionalDataInputString(
    id="protocol",
    label="Analysis Protocol",
    options=["standard", "extended", "rapid"],
    default_value="standard"
)

Integer Input

AdditionalDataInputInteger(
    id="dilution_factor",
    label="Dilution Factor",
    default_value=10,
    options=[1, 5, 10, 50, 100]
)

Float Input

AdditionalDataInputFloat(
    id="concentration",
    label="Sample Concentration (mg/mL)",
    default_value=1.0
)

Boolean Input

AdditionalDataInputBoolean(
    id="requires_cooling",
    label="Requires Cooling",
    default_value=False
)

Conditional Logic in Continuous Loading

Continuous loading supports conditional task execution within each iteration.

continuous_loading = ContinuousLoading(
    id="sample_processing",
    loaded_labware=sample_tubes,
    additional_data_inputs=[
        AdditionalDataInputString(
            id="sample_type",
            label="Sample Type",
            options=["plasma", "serum", "other"]
        )
    ]
)

initial_prep = ActionTask(
    labware_sources=[StoredLabware(sample_tubes, slot=1)],
)

conditional_processing = workflow.If(continuous_loading.out("sample_type") == "plasma").Then(
    plasma_specific_prep := ActionTask(
        labware_sources=[initial_prep.forward(labware=sample_tubes)],
    )
).ElseIf(continuous_loading.out("sample_type") == "serum").Then(
    serum_specific_prep := ActionTask(
        labware_sources=[initial_prep.forward(labware=sample_tubes)],
    )
).Else(
    standard_prep := ActionTask(
        labware_sources=[initial_prep.forward(labware=sample_tubes)],
    )
)

workflow = Workflow(tasks=[continuous_loading, initial_prep, conditional_processing, ...])

Runtime Behavior

During workflow execution:

Operators are presented with a form containing the defined additional data inputs.
Labware items can be loaded dynamically, one at a time.
For each loaded item, the defined task sequence executes with access to the operator-provided data.

Complete Example

sample_processing = ContinuousLoading(
    id="continuous_sample_processing",
    description="Process samples as they arrive",
    loaded_labware=sample_tubes,
    additional_data_inputs=[
        AdditionalDataInputString(
            id="patient_id",
            label="Patient ID"
        ),
        AdditionalDataInputString(
            id="priority",
            label="Priority Level",
            options=["routine", "urgent", "stat"],
            default_value="routine"
        ),
        AdditionalDataInputInteger(
            id="expected_results",
            label="Expected Number of Results",
            default_value=24
        )
    ]
)

scan_samples = ActionTask(
    id="scan_barcodes",
    description="Scan sample barcodes",
    instrument_type="barcode_scanner",
    action="scan",
    time_estimate=30,
    labware_sources=[StoredLabware(sample_tubes, slot=1)]
)

priority_processing = workflow.If(sample_processing.out("priority") == "stat").Then(
    stat_processing := ActionTask(
        id="stat_analysis",
        description="STAT priority analysis",
        instrument_type="analyzer",
        action="stat_analyze",
        time_estimate=180,
        labware_sources=[scan_samples.forward(labware=sample_tubes)]
    )
).Else(
    routine_processing := ActionTask(
        id="routine_analysis",
        description="Routine analysis",
        instrument_type="analyzer",
        action="routine_analyze",
        time_estimate=300,
        labware_sources=[scan_samples.forward(labware=sample_tubes)]
    )
)

workflow = Workflow(
    tasks=[sample_processing, scan_samples, priority_processing, other_tasks...]
)

Predefined Transport Configurations

For complex deployments and multi-workflow setups, you can manage transport configurations at the workcell level using cloud-based transport configs. This allows transport timing and configuration to be versioned, shared across multiple workflows, and updated independently without modifying workflow code.

When to Use Predefined Configs

Use predefined transport configs when:

Your transport layer uses V3 transport with actual measured timings
Multiple workflows share the same workcell and transport infrastructure
Transport configuration requires versioning and change management
Transport settings need to be updated without re-publishing workflows
Your workcell is managed as infrastructure shared across the organization

Use inline transport matrix when:

You have a simple workcell with basic timing estimates
Transport configuration is tightly coupled to a specific workflow
You don’t need versioning or sharing across workflows

Referencing a Transport Config

Reference a cloud-based transport configuration by creating a TransportConfig in your workcell definition:

from linq import Linq
from linq.workcell import Workcell, TransportConfig, Instrument, TransportMatrix

# Reference an existing transport config by ID and type
transport_config = TransportConfig(
    id="550e8400-e29b-41d4-a716-446655440000",  # UUID of the transport config
    type="V3",  # V1 or V3
    version=2,  # Optional: specific version. If omitted, uses latest
)

workcell = Workcell(
    instruments=[...],
    transport_matrix=TransportMatrix(default_transport_time=30),
    transport_config=transport_config,  # Link the config
)

workflow = Workflow(
    workcell=workcell,
    tasks=[...],
    # ... rest of configuration
)

Creating and Managing Configs

Transport configs are typically created and managed via the CLI or SDK client, not in workflow code.

Create a Transport Config

CLI:

linq workcell transport-config-create \
    --name "Lab A Transport V3" \
    --type V3 \
    config.json

SDK:

from linq import Linq

linq = Linq()

config_data = {
    # V3 transport configuration structure
    # (specific format depends on your transport system)
}

result = linq.create_transport_config(
    name="Lab A Transport V3",
    type="V3",
    config=config_data,
)

print(f"Created config with ID: {result.id}")
print(f"Version: {result.version}")

List All Transport Configs

CLI:

# List all configs
linq workcell transport-config-list

# List only V3 configs
linq workcell transport-config-list --type V3

SDK:

# Get all transport configs
all_configs = linq.get_all_transport_configs()

for config in all_configs:
    print(f"ID: {config.id}, Name: {config.name}, Type: {config.type}")

# Get only V3 configs
v3_configs = linq.get_all_transport_configs(transport_config_type="V3")

Get a Specific Transport Config

CLI:

linq workcell transport-config-get \
    --id 550e8400-e29b-41d4-a716-446655440000 \
    --type V3 \
    --version 2

SDK:

# Get the latest version of a config
config = linq.get_transport_config(
    id="550e8400-e29b-41d4-a716-446655440000",
    type="V3",
)

print(f"Config name: {config.name}")
print(f"Current version: {config.version}")

# Get a specific version
config_v1 = linq.get_transport_config(
    id="550e8400-e29b-41d4-a716-446655440000",
    type="V3",
    version=1,
)

Get Version History

CLI:

linq workcell transport-config-versions \
    --id 550e8400-e29b-41d4-a716-446655440000 \
    --type V3

SDK:

versions = linq.get_transport_config_versions(
    id="550e8400-e29b-41d4-a716-446655440000",
    type="V3",
)

for version_num, created_timestamp in versions:
    print(f"Version {version_num}: {created_timestamp}")

Update a Transport Config

Updating a config creates a new version while preserving previous versions.

SDK:

updated_config = linq.update_transport_config(
    id="550e8400-e29b-41d4-a716-446655440000",
    type="V3",
    config={
        # Updated V3 transport configuration
    }
)

print(f"Updated to version: {updated_config.version}")

Rename a Transport Config

CLI:

linq workcell transport-config-rename \
    --id 550e8400-e29b-41d4-a716-446655440000 \
    --type V3 \
    --name "Lab A Transport V3 Updated"

SDK:

renamed_config = linq.rename_transport_config(
    id="550e8400-e29b-41d4-a716-446655440000",
    type="V3",
    name="Lab A Transport V3 Updated",
)

print(f"Renamed to: {renamed_config.name}")

Delete a Transport Config

Warning

Deleting a config will break any workcells that reference it. Unlink the config from all workcells before deletion.

CLI:

linq workcell transport-config-delete \
    --id 550e8400-e29b-41d4-a716-446655440000 \
    --type V3

SDK:

linq.delete_transport_config(
    id="550e8400-e29b-41d4-a716-446655440000",
    type="V3",
)

print("Config deleted")

Linking Config to Workcell

You can associate a transport config with a workcell so that all workflows deployed to that workcell automatically use the linked config. This is useful for standardizing transport behavior across multiple workflows.

Via SDK:

from linq import Linq
from uuid import UUID

linq = Linq()

linq.update_workcell_linked_transport_config(
    workspace_id=UUID("workspace-uuid"),
    workcell_id=UUID("workcell-uuid"),
    transport_config_id=UUID("550e8400-e29b-41d4-a716-446655440000"),
    transport_config_type="V3",
    transport_config_version=2,  # Optional: defaults to latest if not specified
    reason="Updated transport config for improved accuracy"
)

Via CLI:

linq workcell transport-config-set \
    --workspace-id <workspace-uuid> \
    --workcell-id <workcell-uuid> \
    --transport-config-id 550e8400-e29b-41d4-a716-446655440000 \
    --transport-config-type V3 \
    --transport-config-version 2

Once linked, all workflows using that workcell will reference the predefined transport config automatically.

Using Transport Config During Planning

When planning a workflow, you can optionally specify a transport config ID to override the workcell’s default linked config. This allows different planning runs to use different transport configurations for comparison or testing.

SDK:

from linq import Linq
from uuid import UUID

linq = Linq()

# Plan with a specific transport config
plan_status = linq.plan_workflow(
    workflow_id=workflow_id,
    transport_config_id=UUID("550e8400-e29b-41d4-a716-446655440000"),
    parameter_values=[...],
)

print(f"Plan started with status: {plan_status.status}")
print(f"Plan ID: {plan_status.id}")

Note

If transport_config_id is specified during planning, it takes precedence over the workcell’s linked config. This is useful for:

Testing different transport configurations
Comparing plan quality across different configs
Temporarily using a specific config version for a single plan run

CLI:

linq workflow plan start \
    --workflow-id <workflow-id> \
    --transport-config-id 550e8400-e29b-41d4-a716-446655440000

Tasks and Transport

Overview

ActionTask basics

Data Connector

CodeTask

Implicit Actions

Implicit Transport

Controlling Labware Transport Order

Task Dependencies

Expressing Task Dependencies

Task Dependency vs. Labware Dependency

Complex Dependency Graphs

Continuous Loading

Basic Continuous Loading

Adding Tasks to Continuous Loading

Additional Data Input Types

Conditional Logic in Continuous Loading

Runtime Behavior

Complete Example

Predefined Transport Configurations

When to Use Predefined Configs

Referencing a Transport Config

Creating and Managing Configs

Create a Transport Config

List All Transport Configs

Get a Specific Transport Config

Get Version History

Update a Transport Config

Rename a Transport Config

Delete a Transport Config

Linking Config to Workcell

Using Transport Config During Planning

See Also