# Copyright 2026 qBraid
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Module defining Quantinuum device class.
"""
from __future__ import annotations
import asyncio
import functools
import os
import uuid
from collections import OrderedDict
from datetime import datetime, timezone
from typing import TYPE_CHECKING
from qbraid._logging import logger
from qbraid.runtime.device import QuantumDevice
from qbraid.runtime.enums import DeviceStatus
from qbraid.runtime.exceptions import ResourceNotFoundError
from ._transport import (
RETRYABLE_STATUS_CODES,
bounded_int_env,
ensure_bounded_client,
positive_float_env,
retry_transient,
)
from .availability import queue_lengths
from .exceptions import QuantinuumDeviceError
from .job import QuantinuumJob
if TYPE_CHECKING:
from collections.abc import Mapping
from pytket.circuit import Circuit
from qbraid.runtime.profile import TargetProfile
DEFAULT_COMPILE_TIMEOUT_SECONDS = 900.0
DEFAULT_OPT_LEVEL = 1
#: Highest pytket optimisation level NEXUS accepts.
MAX_OPT_LEVEL = 2
[docs]
class QuantinuumDevice(QuantumDevice):
"""Quantinuum NEXUS device interface."""
[docs]
def __init__(self, profile: TargetProfile, **kwargs):
super().__init__(profile=profile, **kwargs)
@property
def backend_info(self):
"""Return the pytket BackendInfo for this device (from the profile)."""
return self.profile.backend_info
def __str__(self):
"""String representation of the QuantinuumDevice object."""
return f"{self.__class__.__name__}('{self.id}')"
def status(self) -> DeviceStatus:
"""Return the current status of the Quantinuum device."""
# The NEXUS machine status endpoint only supports hardware-hosted
# devices; cloud-hosted (Nexus-hosted) emulators are always online.
if getattr(self.profile, "nexus_hosted", False):
return DeviceStatus.ONLINE
# pylint: disable=import-outside-toplevel
import qnexus as qnx
from qnexus.client.devices import DeviceStateEnum
# pylint: enable=import-outside-toplevel
ensure_bounded_client()
cfg = qnx.models.QuantinuumConfig(device_name=self.id)
status = qnx.devices.status(cfg)
if status in (DeviceStateEnum.ONLINE, DeviceStateEnum.RESERVED_ONLINE):
return DeviceStatus.ONLINE
if status in (DeviceStateEnum.MAINTENANCE, DeviceStateEnum.RESERVED_MAINTENANCE):
return DeviceStatus.UNAVAILABLE
return DeviceStatus.OFFLINE
def queue_depth(self) -> int:
"""Return the number of jobs queued on this NEXUS device.
NEXUS reports queue lengths per credential issuer rather than per device, so a
caller refreshing several devices can fetch them in one request with
:func:`~qbraid.runtime.quantinuum.availability.queue_lengths` instead of calling
this once per device.
Raises:
ResourceNotFoundError: NEXUS did not report a queue length for this device,
which is the case for cloud-hosted emulators — they do not queue behind
hardware, so there is no depth to report rather than a depth of zero.
"""
lengths = queue_lengths([self.id])
if self.id not in lengths:
raise ResourceNotFoundError(f"NEXUS reported no queue length for '{self.id}'.")
return lengths[self.id]
def submit( # pylint: disable=arguments-differ
self,
run_input: Circuit | list[Circuit],
shots: int = 1000,
project_name: str | None = None,
optimisation_level: int | None = None,
job_properties: Mapping[str, bool | int | float | str] | None = None,
) -> QuantinuumJob:
"""Compile and submit a pytket circuit (or batch) to the Quantinuum device.
NEXUS requires circuits to be uploaded, compiled, and then executed as
separate job stages. This method blocks while waiting for the
compilation step to finish, then returns a :class:`QuantinuumJob`
referencing the asynchronous execution job.
The ``run_input`` is assumed to already be in pytket ``Circuit`` form;
the qBraid transpiler pipeline (invoked via the base class ``run``)
handles any upstream format conversions based on the device's
:class:`TargetProfile`.
Args:
run_input: pytket ``Circuit`` or a list thereof to execute.
shots: Number of shots per circuit. Defaults to 1000.
project_name: NEXUS project name to scope the compile and execute
jobs under. Falls back to the ``QUANTINUUM_NEXUS_PROJECT_NAME``
environment variable, and ultimately to ``"qbraid"``.
optimisation_level: pytket optimisation level (0-2) passed to the
NEXUS compile stage. Falls back to the
``QUANTINUUM_NEXUS_OPT_LEVEL`` environment variable, and
ultimately to ``1``.
job_properties: Key/value metadata recorded on the NEXUS execute
job. NEXUS indexes these, so jobs sharing a project can still be
filtered by them (``qnx.jobs.get_all(properties=...)``). Each key
must already be defined on the project via
``qnx.projects.add_property``; NEXUS rejects undefined keys.
Every NEXUS request made here is bounded by a per-request HTTP timeout
(``QUANTINUUM_NEXUS_HTTP_TIMEOUT``, seconds, default ``60``), and the
blocking compilation wait is bounded by
``QUANTINUUM_NEXUS_COMPILE_TIMEOUT`` (seconds, default ``900``);
exceeding either raises :class:`QuantinuumDeviceError`.
"""
# pylint: disable=import-outside-toplevel
import httpx
import qnexus as qnx
import qnexus.exceptions as qnx_exc
from qnexus.models.language import Language
# pylint: enable=import-outside-toplevel
circuits = run_input if isinstance(run_input, list) else [run_input]
resolved_project_name = (
project_name
if project_name is not None
else os.getenv("QUANTINUUM_NEXUS_PROJECT_NAME", "qbraid")
)
resolved_opt_level = (
optimisation_level
if optimisation_level is not None
else bounded_int_env("QUANTINUUM_NEXUS_OPT_LEVEL", DEFAULT_OPT_LEVEL, 0, MAX_OPT_LEVEL)
)
compile_timeout = positive_float_env(
"QUANTINUUM_NEXUS_COMPILE_TIMEOUT", DEFAULT_COMPILE_TIMEOUT_SECONDS
)
# Bound every qnexus socket before the first call: the shared client is
# built with ``timeout=None``, so without this any one of the requests
# below can pin the calling thread forever.
ensure_bounded_client()
project = retry_transient(lambda: qnx.projects.get_or_create(name=resolved_project_name))
backend_config = qnx.QuantinuumConfig(device_name=self.id)
def unique(label: str) -> str:
ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S")
return f"qbraid {label} {ts}-{uuid.uuid4().hex[:6]}"
def upload_circuit(index: int, circuit: Circuit):
return qnx.circuits.upload(
name=unique(f"circuit-{index}"), circuit=circuit, project=project
)
# Pre-execute stages are retried on transient connection errors; a repeat
# at worst orphans an upload or compile job, with no execution cost.
circuit_refs = [
retry_transient(functools.partial(upload_circuit, i, c)) for i, c in enumerate(circuits)
]
compile_job = retry_transient(
lambda: qnx.start_compile_job(
programs=circuit_refs,
name=unique("compile"),
optimisation_level=resolved_opt_level,
backend_config=backend_config,
project=project,
)
)
# NOTE: blocking wait during dispatch; compilation time depends on queue and program
# size. The wait is bounded: an unbounded wait_for leaks the calling thread forever
# if the NEXUS compile job hangs, which starves thread pools in server deployments.
# qnexus <=0.42 defaulted this argument to 900s and later releases silently dropped
# the bound to None, which is what made the leak reachable.
logger.info("Waiting for Quantinuum compilation job %s to complete...", compile_job.id)
try:
qnx.jobs.wait_for(compile_job, timeout=compile_timeout)
except asyncio.TimeoutError as err:
raise QuantinuumDeviceError(
f"Quantinuum compilation job did not complete within {compile_timeout:g} "
"seconds. The compile may still be queued on NEXUS; set "
"QUANTINUUM_NEXUS_COMPILE_TIMEOUT (seconds) to wait longer."
) from err
except qnx_exc.JobError as err:
# wait_for raises this when the compile errors, is cancelled, or is
# terminated. Left unwrapped it escapes submit() as a bare qnexus
# exception, which callers cannot reasonably be asked to catch.
raise QuantinuumDeviceError(
f"Quantinuum compilation job {compile_job.id} did not succeed: {err}"
) from err
# ``get_output`` is lazy and issues its own NEXUS request, so the whole
# fetch stage goes inside the retry, not just the results listing.
compiled_refs = retry_transient(
lambda: [item.get_output() for item in qnx.jobs.results(compile_job)]
)
# Deliberately NOT retried: a disconnect after the server accepted this
# request would double-submit (and double-bill) the execution. But it
# must not go unwrapped either: since the shared client now carries a
# read timeout, a slow NEXUS response here raises after the server may
# already have accepted a billable job, and a bare httpx exception gives
# the caller no way to find it. Name the job first so the error can say
# exactly what to look for. The same ambiguity applies to a gateway
# 502/503/504: qnexus reports those as ``ResourceCreateFailed`` rather
# than an httpx error, and a gateway can fail after forwarding the
# request, so they get the same guidance. Other status codes mean NEXUS
# itself rejected the dispatch, and those propagate untouched.
execute_name = unique("execute")
try:
execute_job = qnx.start_execute_job(
programs=compiled_refs,
name=execute_name,
n_shots=[shots] * len(compiled_refs),
backend_config=backend_config,
project=project,
language=Language.QIR,
properties=OrderedDict(job_properties) if job_properties else None,
)
except (httpx.TransportError, ConnectionError) as err:
raise QuantinuumDeviceError(
f"No response received for the execute dispatch ({err}). NEXUS may still "
f"have accepted the job: check project {resolved_project_name!r} for a job "
f"named {execute_name!r} and cancel it if it is running, or its shots will "
"be billed with no local handle to poll."
) from err
except qnx_exc.ResourceCreateFailed as err:
if err.status_code not in RETRYABLE_STATUS_CODES:
raise
raise QuantinuumDeviceError(
f"Gateway error on the execute dispatch ({err}). NEXUS may still have "
f"accepted the job behind the gateway: check project "
f"{resolved_project_name!r} for a job named {execute_name!r} and cancel "
"it if it is running, or its shots will be billed with no local handle "
"to poll."
) from err
return QuantinuumJob(job_id=str(execute_job.id), device=self, job=execute_job)