Running kernels on a device

bloqade.core.device provides the pieces needed to submit kernels to a backend, wait for them to finish, and read their shot results. The canonical workflow constructs workloads with TaskBuilder and submits them through Device.

TaskBuilder ──Device.run_async()──▶ Future ──result()──▶ Result

Concepts

TaskBuilder. An append-only description of a workload. Each add_subtask call records a kernel invocation, shot count, argument set, and optional user metadata. A builder can mix distinct kernels and can reuse one kernel across multiple subtasks. See Building tasks with TaskBuilder for the complete construction API.

Device. Owns the backend context and the language, serializer, dialects, and validation rules accepted by that backend. Calling Device.run_async finalizes a builder. With dry_run=True, it validates and prints a summary without authenticating or submitting. With dry_run=False, it submits the finalized definition and returns a Future.

Future. A handle to a submitted task. It polls task status against the backend, fetches available shot results into storage, and constructs a Result view. future.result(timeout=...) blocks until the task completes successfully, then fetches all shots and returns a Result. If the task is cancelled, fails, or hits a payload processing error, result() raises instead of returning. A future can also be reattached to an existing task ID via Future.from_task_id or Future.from_storage, which is how you resume from a previous session.

Result. A view over the shots in storage scoped to one or more task IDs and a frame type. result.shot_results() returns one boolean np.ndarray per subtask. The where_subtasks, where_arguments, where_metadata, and where_shots methods narrow the view without copying data.

The canonical `TaskBuilder` flow

The example below submits a simple Squin kernel using TaskBuilder. The context_name sets the context as specified in your ~/.qsh/config.json.

The kernel languages used on this page (squin, qasm2) ship with the bloqade-circuit package, not with bloqade-core itself. Install the relevant extras (for example pip install bloqade-circuit[qasm2] for the QASM2 specialization below) before running these snippets.

from bloqade import squin
from bloqade.core.device import Device, TaskBuilder

# 1. Define a kernel.
@squin.kernel
def bell():
    q = squin.qalloc(2)
    squin.h(q[0])
    squin.cx(q[0], q[1])

# 2. Configure the device and build its workload.
device = Device(
    context_name="my-context",
    program_language="squin",
    language_version="0.1.0",
)
builder = TaskBuilder()
builder.add_subtask(
    bell,
    num_shots=2,
    metadata={"tag": "bell"},      # metadata is arbitrary user JSON
)

# 3a. Dry-run first to see what would be submitted.
device.run_async(builder, dry_run=True)

# 3b. Submit. The first call triggers browser-based authentication.
future = device.run_async(builder, dry_run=False)

# 4. Wait for completion and read shot bitstrings.
result = future.result(timeout=80.0)
print(result.shot_results())

Each call to builder.add_subtask(...) appends another invocation. This single API covers one kernel, multiple kernels, parameter sweeps, and interleaved workloads. The builder remains editable after a dry run or submission.

The repository includes runnable TaskBuilder examples with a device-level QASM2 serializer. Each script requires bloqade-circuit[qasm2]; replace its context_name with one that matches your local QLAM config.

demo/qasm_single_task.py — submits a QASM2 kernel with the default in-memory storage.
demo/qasm_single_task_persistent.py — submits the same workload with SQLiteStorage so results survive across sessions.

Deprecated task-object workflow

Deprecated

Device.task(), Device.batch_task(), and Device.parameter_scan() are deprecated. They remain available for backwards compatibility, but new code should use TaskBuilder and Device.run_async().

The deprecated task factories and task-owned run_async() method work as follows:

task = device.task(
    kernel=bell,
    num_shots=2,
    metadata={"tag": "bell"},
    program_language="squin",
)

task.run_async(dry_run=True)
future = task.run_async(dry_run=False)
result = future.result(timeout=80.0)

Existing code may continue to use Device.task, Device.batch_task, and Device.parameter_scan. Do not use this workflow for new integrations.

Groups

QLAM groups organize task definitions and control who can access them. The group submitted with a task is resolved in this order, mirroring the qsh CLI:

a task-level group name,
the ~/.qsh config group (plugins.tasks.group, then defaults.group),
otherwise the group is omitted and the server decides according to its configured submission mode.

group = "qec-experiments"

builder = TaskBuilder()
builder.add_subtask(bell, num_shots=100)
future = device.run_async(builder, dry_run=False, group=group)

Submission-level and configured group values are resolved through QLAM before submission. The resolved group ID is recorded on task definitions, including definitions stored in DictStorage and SQLiteStorage. The group a task actually landed in is reported on every task response (future.get_task().group).

Group membership is assigned by your tenant admin. Group administration (including discovery) is a qlam-core/qsh concern and is not exposed by bloqade.

Persisting results

Device.run_async accepts a storage argument that controls where the task definition and fetched shots are kept. If you omit it, a fresh DictStorage is used — that is, the data lives in memory only and is lost when the process exits. Pass a SQLiteStorage instead to keep everything on disk and resume work in a later session:

from bloqade.core.device import SQLiteStorage

storage = SQLiteStorage("results.sql")
future = device.run_async(builder, dry_run=False, storage=storage)
result = future.result(timeout=80.0)

storage.close()  # optional; GC will also close it

In a later session, reattach to the same task without resubmitting:

from bloqade.core.device import Future, SQLiteStorage

storage = SQLiteStorage("results.sql")
future = Future.from_storage(storage=storage, context_name="my-context")
result = future.result(timeout=80.0)

When more than one task ID is present in the storage, pass task_id=... to disambiguate. To fetch a task that you do not have stored yet, use Future.from_task_id(task_id=..., storage=storage, context_name=...) — this fetches the task definition from the backend into storage and returns a future you can drive normally.

Filtering results

A Result is a view over storage rather than a materialized list of shots. Each where_* method returns a new, narrower view by combining the current ShotFilter with the matches from a predicate. Nothing is copied, so the views are cheap and can be chained. Each method takes a Callable[[X], bool] predicate whose input type X matches the row kind being filtered:

where_subtasks — Callable[[dict], bool]. Predicate receives the full subtask record (program index, num_shots, arguments, metadata, …).
where_arguments — Callable[[dict | None], bool]. Predicate receives the per-subtask arguments dict only (or None). Convenient for slicing a parameter scan.
where_metadata — Callable[[dict | None], bool]. Predicate receives the JSON-decoded user metadata you attached when building the task (or None).
where_shots — Callable[[ShotResult], bool]. Predicate receives individual ShotResult rows; the optional predicate_filter: ShotFilter | None argument controls which shots the predicate is evaluated against (see frame types below).

# Suppose `result` came from a parameter scan whose arguments look like
#   {"theta": float, "phi": float, "lam": float}
# and whose per-subtask metadata looks like
#   {"recheck": bool, ...}

def lam_is_zero(arguments: dict | None) -> bool:
    return arguments is not None and arguments.get("lam") == 0.0

def needs_recheck(metadata: dict | None) -> bool:
    return metadata is not None and metadata.get("recheck", False)

# Slice the scan to lam == 0, then keep only subtasks tagged for re-analysis.
narrowed = result.where_arguments(lam_is_zero).where_metadata(needs_recheck)
print(narrowed.shot_results())

Frame types

Every shot is recorded under one of two frame types:

DETECTED — the final measurement at the end of the program. This is the default scope of a Result and what shot_results() returns.
SORTED — the readout taken right after atom sorting. Useful as a precondition to check whether sorting was successful, so you can discard the corresponding DETECTED shot when it was not.

where_shots makes this precondition explicit by accepting a predicate_filter. The predicate is evaluated against the shots selected by that filter, but the returned view keeps the original frame scope:

from bloqade.core.device import ShotFilter, ShotResult

def atom_sorting_was_successful(shot: ShotResult) -> bool:
    return bool(shot.bitstring.all())

clean = result.where_shots(
    atom_sorting_was_successful,
    predicate_filter=ShotFilter(frame_type="SORTED"),
)
print(clean.shot_results())  # DETECTED shots whose SORTED frame passed

The ShotFilter and StorageFilter classes in local_storage.py describe every field you can scope on directly if you need finer control than the where_* helpers provide.

Specializing deprecated task objects

This section documents customization of the deprecated task-object workflow for maintainers of existing integrations. New code should configure serialization and validation on the device and submit a TaskBuilder through Device.run_async().

The default legacy task serializes kernels with kirin's JSON encoder, which fits most use cases. When the backend expects a different program format, or when you want to attach extra fields to the submitted task, subclass the relevant task type and point the device at it.

As an example, consider a backend that expects QASM2 source tagged as version 2.0.0. The default SingleKernelTask would encode the kernel as kirin JSON, which the backend cannot parse. Overriding serialize_kernel switches the wire format, and overriding program_language_version records the right version. Pointing a Device subclass at the new task is then enough to change everything downstream:

from dataclasses import dataclass, field
from bloqade.qasm2.emit import QASM2
from kirin.ir.method import Method
from bloqade.core.device import Device, Future, Result
from bloqade.core.device.task import SingleKernelTask

@dataclass
class QASM2Task(SingleKernelTask):
    @property
    def program_language_version(self) -> str:
        return "2.0.0"
    def serialize_kernel(self, kernel: Method) -> str:
        return QASM2().emit_str(kernel)

@dataclass
class QASM2Device(Device):
    single_kernel_task_cls: type[SingleKernelTask[Future[Result]]] = field(
        default=QASM2Task, init=False
    )

A Device holds a class reference for each task shape it can build (single_kernel_task_cls, kernel_batch_task_cls, parameter_scan_task_cls). Override the ones you need; the device's task, batch_task, and parameter_scan methods will instantiate your subclass. The task records its language on the definition as <program_language>.v<program_language_version> — here "qasm.v2.0.0", since QASM2Task pins the version to "2.0.0". program_language_version defaults to the language_version argument (which must be a semantic version); override the property, as above, when a subclass should fix or compute the version. Set program_language to whatever the backend expects:

from bloqade import qasm2

@qasm2.main
def bell():
    q = qasm2.qreg(2)
    qasm2.h(q[0])
    qasm2.cx(q[0], q[1])

device = QASM2Device(context_name="my-qasm-context")
task = device.task(            # returns a QASM2Task instance
    kernel=bell,
    num_shots=2,
    metadata={"tag": "bell"},
    program_language="qasm",
)
future = task.run_async(dry_run=False)
result = future.result(timeout=80.0)

Other common extension points on the task itself are program_language_version (the version recorded on the task definition and passed to the default Kirin serializer when encoding the kernel — override it as QASM2Task does above, or pass language_version=... to device.task(...) for a static version), summary (text printed on dry-run), and create_task_definition (full control over the submitted payload).

The same pattern extends to the Future class. Device.future_cls is the class the task instantiates after submission, so setting it to a Future subclass on your device lets you attach backend-specific helpers — extra polling logic, logical-result decoding, custom export hooks — without touching the task or device flow. Future.result_cls plays the same role for Result, so a specialized Future can surface a specialized result view through future.result(...).

Logging

Logging is opt-in and off by default. When enabled, task submissions and status fetches are logged at INFO level via loguru to a file called bloqade.log in the current working directory.

Turn it on programmatically with set_logging. Call it once to opt in, optionally choosing the file and level:

from bloqade.core.device import set_logging

set_logging()                                  # bloqade.log, INFO level
# or, with a custom file and level:
set_logging(path="run.log", level="DEBUG")

set_logging(enabled=False)                     # turn logging back off

Alternatively, set the BLOQADE_LOGGING environment variable to "1" before importing bloqade.core.device to enable the default file sink at import time:

export BLOQADE_LOGGING=1

If the log file cannot be created (for example, the path is not writable), set_logging emits a RuntimeWarning and leaves logging disabled rather than raising.