# 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 job class.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from qbraid._logging import logger
from qbraid.runtime.enums import JobStatus
from qbraid.runtime.job import QuantumJob
from qbraid.runtime.result import Result
from qbraid.runtime.result_data import GateModelResultData
from ._transport import ensure_bounded_client
from .exceptions import QuantinuumJobError
if TYPE_CHECKING:
from qbraid_core.services.runtime.schemas.job import Program
from qnexus.models.job_status import JobStatus as NexusJobStatus
from qnexus.models.references import ExecuteJobRef, ExecutionProgram
from qbraid.runtime.quantinuum.device import QuantinuumDevice
#: Every member of ``qnexus.models.job_status.JobStatusEnum``, keyed by value so
#: the module does not need qnexus imported to be defined. An unmapped status
#: raises rather than degrading to ``UNKNOWN``: ``UNKNOWN`` is not terminal, so
#: a new NEXUS state that actually stops a job would otherwise be polled until
#: the caller's own timeout, silently.
#:
#: - ``SUBMITTED`` is the state a job enters the moment NEXUS accepts it, before
#: the device queues it, so it is the first status most jobs ever report.
#: - ``RETRYING`` means NEXUS is re-attempting the job; it has produced no
#: results, so it maps to ``QUEUED`` rather than ``RUNNING``.
#: - ``TERMINATED`` (stopped by the platform) and ``DEPLETED`` (credits
#: exhausted) are terminal failures.
_QUANTINUUM_STATUS_MAP: dict[str, JobStatus] = {
"COMPLETED": JobStatus.COMPLETED,
"ERROR": JobStatus.FAILED,
"TERMINATED": JobStatus.FAILED,
"DEPLETED": JobStatus.FAILED,
"CANCELLED": JobStatus.CANCELLED,
"CANCELLING": JobStatus.CANCELLING,
"SUBMITTED": JobStatus.QUEUED,
"QUEUED": JobStatus.QUEUED,
"RETRYING": JobStatus.QUEUED,
"RUNNING": JobStatus.RUNNING,
}
def _map_quantinuum_status(last_status: str | None) -> JobStatus:
"""Map a qnexus job status value to a qBraid :class:`JobStatus`.
Raises:
QuantinuumJobError: If NEXUS reports a status the map does not cover.
"""
if last_status is None:
return JobStatus.UNKNOWN
try:
return _QUANTINUUM_STATUS_MAP[last_status]
except KeyError as err:
raise QuantinuumJobError(f"Unrecognized Quantinuum job status {last_status!r}.") from err
def _program_from_input_ref(ref: ExecutionProgram) -> Program | None:
"""Convert a NEXUS execution input reference to a qBraid :class:`Program`.
NEXUS decides which arm of ``ExecutionProgram`` a job returns, so this
branches on the concrete type rather than assuming one. Returns ``None`` for
an arm with no ``Program.format`` counterpart (HUGR) rather than guessing at
one; the caller keeps that ``None`` in place to stay aligned with the batch.
The compiled H-series circuits use native gates (``PhasedX``, ``ZZPhase``),
which the default ``qelib1`` export cannot represent — hence ``hqslib1``.
The emitted QASM carries ``include "hqslib1.inc";``, which pyqasm cannot
parse today; that affects neither storing nor serving it.
"""
# pylint: disable-next=import-outside-toplevel
from qbraid_core.services.runtime.schemas.job import Program
# pylint: disable-next=import-outside-toplevel
from qnexus.models.references import CircuitRef, QIRRef
if isinstance(ref, CircuitRef):
# pylint: disable-next=import-outside-toplevel
from pytket.qasm import circuit_to_qasm_str
# pylint: disable-next=import-outside-toplevel
from qbraid.transpiler.conversions.pytket._utils import QASM_MAXWIDTH
return Program(
format="qasm2",
data=circuit_to_qasm_str(ref.download_circuit(), "hqslib1", maxwidth=QASM_MAXWIDTH),
)
if isinstance(ref, QIRRef):
return Program(format="qir.bc", data=ref.download_qir())
logger.debug("No compiled-program format for NEXUS input ref %s", type(ref).__name__)
return None
[docs]
class QuantinuumJob(QuantumJob):
"""Quantinuum NEXUS job class."""
[docs]
def __init__(
self,
job_id: str,
device: QuantinuumDevice | None = None,
job: ExecuteJobRef | None = None,
**kwargs: Any,
) -> None:
super().__init__(job_id=job_id, device=device, **kwargs)
self._job = job
self._status_detail: NexusJobStatus | None = None
self._input_refs: list[ExecutionProgram] | None = None
self._compiled_programs: list[Program | None] | None = None
def _get_ref(self) -> ExecuteJobRef:
"""Return the cached qnexus job reference, or look it up by ID."""
if self._job is not None:
return self._job
try:
# pylint: disable-next=import-outside-toplevel
import qnexus as qnx
ensure_bounded_client()
self._job = qnx.jobs.get(id=self.id)
return self._job
except Exception as exc:
raise QuantinuumJobError(f"Unable to retrieve Quantinuum job {self.id}") from exc
def status(self) -> JobStatus:
"""Return the current status of the Quantinuum job.
Queries NEXUS for a fresh status rather than reading ``last_status``
off the job reference. That field is a plain snapshot taken when the
reference was built, so a job object held across a run -- the ordinary
``job = device.run(...)`` case -- would report its submission-time
status forever and never observe completion.
"""
if self._cache_metadata.get("status") in JobStatus.terminal_states():
return self._cache_metadata["status"]
# pylint: disable-next=import-outside-toplevel
import qnexus as qnx
try:
ref = self._get_ref()
ensure_bounded_client()
detail = qnx.jobs.status(ref)
except Exception as exc:
raise QuantinuumJobError(f"Unable to retrieve job status for {self.id}") from exc
status = _map_quantinuum_status(detail.status.value)
if status == JobStatus.FAILED:
logger.error(
"Quantinuum job %s failed: %s",
self.id,
detail.error_detail or detail.message or "no error message",
)
self._status_detail = detail
self._cache_metadata["status"] = status
return status
def _get_status_detail(self) -> NexusJobStatus:
"""Return the NEXUS status detail, fetching it when not already cached.
:meth:`status` early-returns on a cached terminal status without
touching NEXUS, so a job constructed with ``status=COMPLETED`` (or one
rehydrated from persisted metadata) reaches the timing and error-detail
readers with ``_status_detail`` still unset. Fetching on demand keeps
those readers working for such jobs instead of failing on a job that
genuinely finished.
"""
if self._status_detail is not None:
return self._status_detail
# pylint: disable-next=import-outside-toplevel
import qnexus as qnx
try:
ref = self._get_ref()
ensure_bounded_client()
self._status_detail = qnx.jobs.status(ref)
except Exception as exc:
raise QuantinuumJobError(f"Unable to retrieve job status for {self.id}") from exc
return self._status_detail
def _resolve_device_id(self, ref: ExecuteJobRef) -> str:
"""Resolve the target device name for a job.
Prefers the device this job was constructed with, then tries to read
the device name from the NEXUS job's ``backend_config`` when it is a
:class:`~quantinuum_schemas.models.backend_config.QuantinuumConfig`
(other ``BackendConfig`` subclasses do not expose ``device_name``).
Falls back to the generic ``"quantinuum"`` label if neither is set.
"""
# pylint: disable-next=import-outside-toplevel
from quantinuum_schemas.models.backend_config import QuantinuumConfig
if self._device is not None:
return self._device.id
backend_config = ref.backend_config
if isinstance(backend_config, QuantinuumConfig):
return backend_config.device_name
return "quantinuum"
def cancel(self) -> None:
"""Cancel the Quantinuum job."""
# pylint: disable-next=import-outside-toplevel
import qnexus as qnx
try:
ensure_bounded_client()
qnx.jobs.cancel(self._get_ref())
except Exception as exc:
raise QuantinuumJobError(f"Failed to cancel Quantinuum job {self.id}") from exc
def execution_time_s(self) -> float | None:
"""Return the wall-clock execution time of the job in seconds.
Computed from the NEXUS status timestamps
(``completed_time - running_time``). This duration reflects the total
time the job spent in the "running" phase on the service and may
include queueing, calibration, and other backend checks in addition
to on-device execution. qnexus does not currently expose a more
granular, on-device-only metric.
The timestamps come from the status fetched by :meth:`status`, not from
the job reference's snapshot, which is fixed at the moment the
reference was built and so carries no completion time at all.
Returns:
The execution time in seconds, or ``None`` if the job has not
completed.
Raises:
QuantinuumJobError: If the job is completed but timing details
are unavailable.
"""
if self.status() != JobStatus.COMPLETED:
return None
detail = self._get_status_detail()
completed_time = detail.completed_time
running_time = detail.running_time
if completed_time is None or running_time is None:
raise QuantinuumJobError(
f"Execution time not available for {self.id}: "
"completed_time or running_time is missing"
)
return (completed_time - running_time).total_seconds()
def result(self) -> Result[GateModelResultData]:
"""Return the result of the Quantinuum job."""
# pylint: disable-next=import-outside-toplevel
import qnexus as qnx
# pylint: disable-next=import-outside-toplevel
from pytket.circuit import BasisOrder
self.wait_for_final_state()
if self.status() == JobStatus.FAILED:
detail = self._status_detail
message = (
(detail.error_detail or detail.message) if detail is not None else None
) or "no error message"
raise QuantinuumJobError(f"Quantinuum job {self.id} failed: {message}")
ref = self._get_ref()
try:
ensure_bounded_client()
results = qnx.jobs.results(ref)
except Exception as exc:
raise QuantinuumJobError(
f"Failed to fetch results for Quantinuum job {self.id}"
) from exc
if not results:
raise QuantinuumJobError(f"No results available for Quantinuum job {self.id}")
# Quantinuum / pytket use least-significant-bit-first ordering by default.
# Convert to most-significant-bit-first (dlo = descending lexicographic order)
# for consistency with other qBraid providers.
# ``download_result`` is lazy and issues its own NEXUS request per item,
# so the loop needs the same wrapping as the results listing above.
all_counts: list[dict[str, int]] = []
input_refs: list[ExecutionProgram] | None = []
try:
for result_item in results:
counts = result_item.download_result().get_counts(basis=BasisOrder.dlo)
all_counts.append({"".join(map(str, k)): v for k, v in counts.items()})
# Served from the fetch ``download_result`` just made: both read the
# same cached payload on the ref, so this costs no extra request.
#
# Guarded separately: the compiled program is optional metadata, and
# letting its failure escape would discard measurement counts that
# were already downloaded successfully. Dropping to None rather than
# a partial list leaves compiled_program() free to refetch later.
if input_refs is not None:
try:
input_refs.append(result_item.get_input())
except Exception: # pylint: disable=broad-exception-caught
logger.debug(
"Could not capture the compiled-program reference for "
"Quantinuum job %s",
self.id,
exc_info=True,
)
input_refs = None
except Exception as exc:
raise QuantinuumJobError(
f"Failed to download results for Quantinuum job {self.id}"
) from exc
self._input_refs = input_refs
measurement_counts = all_counts[0] if len(all_counts) == 1 else all_counts
device_id = self._resolve_device_id(ref)
return Result[GateModelResultData](
device_id=device_id,
job_id=self.id,
success=True,
data=GateModelResultData(measurement_counts=measurement_counts),
)
def compiled_program(self) -> Program | list[Program | None] | None:
"""Return the NEXUS-compiled program that executed on the QPU.
Reveals the native gates and qubit mapping Quantinuum's compiler chose —
information not recoverable from measurement counts. ``format`` is
``"qasm2"`` (``hqslib1`` dialect) for circuit jobs and ``"qir.bc"`` for
QIR jobs. A batch submission returns a list with one entry per program,
positionally aligned, holding ``None`` for any program type qBraid
has no format for.
``None`` when the job has no compiled program to report: it has not
completed, or NEXUS returned only program types qBraid has no format for.
Absence before completion is not cached, so a job polled early picks its
program up on a later call.
Downloading the program itself is a NEXUS request, so it is made on first
call and cached. Calling :meth:`result` first is cheaper overall — it
captures the references for free — but is not required.
"""
if self._compiled_programs is None:
if self._input_refs is None:
self._load_input_refs()
if self._input_refs is None:
# Unfinished job: absence is not cached, so a call after the job
# completes still picks the program up. Caching ``[]`` here would
# make every later call return ``None`` for the job's whole life.
return None
try:
# One entry per submitted program, ``None`` where NEXUS returned a
# type with no ``Program.format``. Dropping those instead would
# shorten the list and silently misalign it with the batch.
self._compiled_programs = [_program_from_input_ref(ref) for ref in self._input_refs]
except Exception as exc:
raise QuantinuumJobError(
f"Failed to download compiled program for Quantinuum job {self.id}"
) from exc
if not any(program is not None for program in self._compiled_programs):
return None
if len(self._compiled_programs) == 1:
return self._compiled_programs[0]
return self._compiled_programs
def _load_input_refs(self) -> None:
"""Populate ``_input_refs`` for a job whose results were never downloaded."""
# pylint: disable-next=import-outside-toplevel
import qnexus as qnx
if self.status() != JobStatus.COMPLETED:
return
try:
ensure_bounded_client()
self._input_refs = [item.get_input() for item in qnx.jobs.results(self._get_ref())]
except Exception as exc:
raise QuantinuumJobError(
f"Failed to fetch compiled program for Quantinuum job {self.id}"
) from exc