Workflow Engine Quickstart

This guide helps you get started with the QDash workflow engine.

Overview

The workflow engine executes calibration tasks using Prefect. For detailed architecture, see Engine Architecture.

Key Components

workflow/
├── engine/           # Core execution engine
│   ├── orchestrator.py   # Session lifecycle (start here)
│   ├── task_runner.py    # Prefect task wrappers
│   └── task/             # Task execution layer
├── calibtasks/       # Calibration task definitions
└── service/          # High-level service APIs

Basic Usage

1. Running a Calibration Session

from qdash.workflow.engine import CalibOrchestrator, CalibConfig

# Configure the session
config = CalibConfig(
    username="alice",
    project_id="proj-1",
    chip_id="64Qv3",
    qids=["0", "1"],
    execution_id="20240101-001",
)

# Initialize and run
orchestrator = CalibOrchestrator(config)
orchestrator.initialize()

# Execute a task
result = orchestrator.run_task("CheckRabi", qid="0")

# Complete the session
orchestrator.complete()

2. Understanding the Execution Flow

1. CalibOrchestrator.initialize() - Creates directories, connects to backend
2. orchestrator.run_task() - Executes a single calibration task
3. orchestrator.complete() - Finalizes session, saves results to MongoDB

Writing Flow Templates

Flow templates are user-editable recipes. Keep each step's task names and execution order visible in the template file itself.

• Define task lists as explicit lists of task-name strings in the template, or pass a literal list directly to tasks=. Do not import task lists from other templates or rely on a calibration step's default task list.
• Add short comments explaining each stage, repeated calibration round, filter, and hardware configuration checkpoint.
• When a combined template reproduces standalone stages, keep the task lists explicit in both places and test that their contents and order agree.
• Preserve intentional repetitions: repeated Rabi checks and readout optimization rounds should remain visible so users can review and edit them.

one_qubit.py lists its coarse and fine tasks locally, with a status filter between them. Its tests compare those lists with coarse_one.py and fine_one.py.

Adding a New Calibration Task

1. Create a task class in workflow/calibtasks/:

from qdash.workflow.calibtasks.base import BaseTask

class MyNewTask(BaseTask):
    name = "MyNewTask"
    
    def preprocess(self, backend):
        # Extract input parameters
        pass
    
    def run(self, backend):
        # Execute measurement
        pass
    
    def postprocess(self, backend, result):
        # Process results, generate figures
        pass

2. Register it in workflow/calibtasks/active_protocols.py

Testing

Use the FakeBackend for testing without hardware:

config = CalibConfig(
    backend_name="fake",  # Use fake backend
    project_id="proj-1",
    ...
)

See Testing Guidelines for more details.

Cancellation Support

All top-level @flow decorators must register the on_flow_cancellation hook to support cancellation from the UI:

from qdash.workflow.service.calib_service import on_flow_cancellation

@flow(on_cancellation=[on_flow_cancellation])
def my_calibration_flow(
    username: str,
    chip_id: str,
    project_id: str | None = None,
    flow_name: str | None = None,
    ...
):
    cal = CalibService(username, chip_id, ...)
    try:
        # ... run tasks ...
        cal.finish_calibration()
    except BaseException as e:
        from qdash.workflow.service.calib_service import _is_cancellation
        if _is_cancellation(e):
            cal.cancel_calibration()
        else:
            cal.fail_calibration(str(e))
        raise

When cancelled, Prefect kills the process with SIGTERM and runs the on_cancellation hook. The hook updates the execution and task statuses to cancelled and releases the execution lock.

See Engine Architecture — Cancellation for implementation details.

Next Steps

• Engine Architecture - Deep dive into components
• Testing Guidelines - How to test workflow code