feat: ability to show incomplete job items (#271)

Author: vanyae-cqc

.cz.toml

@@ -2,7 +2,7 @@
 name = "cz_customize"
 tag_format = "v$version"
 version_scheme = "semver"
-version = "0.33.0"
+version = "0.34.0"
 version_files = ["pyproject.toml"]
 update_changelog_on_bump = true
 major_version_zero = true

.gitignore

@@ -14,6 +14,7 @@ __pycache__/
 
 dist/
 docs/_build
+.coverage
 
 # IPython
 .ipynb_checkpoints

CHANGELOG.md

@@ -2,6 +2,14 @@
 
 # `qnexus` Release Notes
 
+## v0.34.0 (2025-09-30)
+
+
+### Added
+
+- Return `IncompleteJobItemRef` for job items that haven't completed when fetching job results.
+
+
 ## v0.33.0 (2025-09-29)
 
 ### Removed

examples/basics/jobs_results.ipynb

@@ -194,7 +194,9 @@
     "\n",
     "By default, you can only retrieve results for jobs with the COMPLETED status (meaning that all items in the Job have successfully completed). \n",
     "\n",
-    "In some contexts you may want to retrieve results for the completed items in an otherwise pending or errored job. For example, maybe you have submitted 10 circuits to be executed on quantum hardware, but only 6 of them have completed before you ran out of credit quota for that device. In this case you can still get the results from the completed items."
+    "In some contexts you may want to retrieve results for the completed items in an otherwise pending or errored job. For example, maybe you have submitted 10 circuits to be executed on quantum hardware, but only 6 of them have completed before you ran out of credit quota for that device. In this case you can still get the results from the completed items.\n",
+    "\n",
+    "Any items that have not completed will be returned as an `IncompleteJobItemRef` with an individual status."
    ]
   },
   {

integration/test_backend_configs.py

@@ -7,7 +7,11 @@
 
 import qnexus as qnx
 from qnexus.models.job_status import JobStatusEnum
-from qnexus.models.references import ProjectRef
+from qnexus.models.references import (
+    CompilationResultRef,
+    ExecutionResultRef,
+    ProjectRef,
+)
 
 CONFIGS_REQUIRE_NO_MEASURE = [qnx.AerUnitaryConfig]
 CONFIGS_NOT_TO_EXECUTE = [
@@ -53,7 +57,11 @@ def test_basic_backend_config_usage(
         ).backend_config.model_dump(exclude={"noisy_simulation"})
 
         execute_job_ref = qnx.start_execute_job(
-            programs=[item.get_output() for item in qnx.jobs.results(compile_job_ref)],
+            programs=[
+                item.get_output()
+                for item in qnx.jobs.results(compile_job_ref)
+                if isinstance(item, CompilationResultRef)
+            ],
             name=f"execute job for {test_case_name}",
             n_shots=[100],
             backend_config=backend_config,
@@ -71,4 +79,5 @@ def test_basic_backend_config_usage(
         execute_job_result_refs = qnx.jobs.results(execute_job_ref)
 
         for result_ref in execute_job_result_refs:
+            assert isinstance(result_ref, ExecutionResultRef)
             assert isinstance(result_ref.download_result(), BackendResult)

integration/test_jobs.py

@@ -4,6 +4,7 @@
 from datetime import datetime
 from time import sleep
 from typing import Any, Callable, ContextManager
+from uuid import UUID
 
 import pandas as pd
 import pytest
@@ -22,6 +23,8 @@
     CompilationResultRef,
     CompileJobRef,
     ExecuteJobRef,
+    ExecutionResultRef,
+    IncompleteJobItemRef,
     JobRef,
     Ref,
 )
@@ -205,13 +208,14 @@ def test_submit_compile(
         compile_results = qnx.jobs.results(compile_job_ref)
 
         assert len(compile_results) == 1
-        assert isinstance(compile_results[0], CompilationResultRef)
-        test_ref_serialisation("compile_result", compile_results[0])
+        compilation_result = compile_results[0]
+        assert isinstance(compilation_result, CompilationResultRef)
+        test_ref_serialisation("compile_result", compilation_result)
 
-        assert isinstance(compile_results[0].get_input(), CircuitRef)
-        assert isinstance(compile_results[0].get_output(), CircuitRef)
+        assert isinstance(compilation_result.get_input(), CircuitRef)
+        assert isinstance(compilation_result.get_output(), CircuitRef)
 
-        first_pass_data = compile_results[0].get_passes()[0]
+        first_pass_data = compilation_result.get_passes()[0]
 
         assert isinstance(first_pass_data, CompilationPassRef)
         assert isinstance(first_pass_data.get_input(), CircuitRef)
@@ -287,11 +291,25 @@ def test_get_results_for_incomplete_compile(
                 project=proj_ref,
                 backend_config=qnx.AerConfig(),
             )
-
+        # check the job while not complete
         assert isinstance(compile_job_ref, CompileJobRef)
         assert qnx.jobs.status(compile_job_ref).status != JobStatusEnum.COMPLETED
+
+        with pytest.raises(qnx_exc.ResourceFetchFailed):
+            qnx.jobs.results(compile_job_ref)
+
         compile_results = qnx.jobs.results(compile_job_ref, allow_incomplete=True)
-        assert len(compile_results) == 0
+        assert len(compile_results) == 1
+        compile_item = compile_results[0]
+        assert isinstance(compile_item, IncompleteJobItemRef)
+        assert isinstance(compile_item.job_item_integer_id, int)
+        assert compile_item.last_status != JobStatusEnum.COMPLETED
+
+        # check the job after completion
+        qnx.jobs.wait_for(compile_job_ref)
+        complete_results = qnx.jobs.results(compile_job_ref, allow_incomplete=True)
+        assert len(complete_results) == 1
+        assert isinstance(complete_results[0], CompilationResultRef)
 
 
 def test_compile_hypertket(
@@ -348,6 +366,7 @@ def test_submit_execute(
         assert len(execute_results) == 1
 
         first_result = execute_results[0]
+        assert isinstance(first_result, ExecutionResultRef)
         assert isinstance(first_result.get_input(), CircuitRef)
         assert isinstance(first_result.download_result(), BackendResult)
         assert isinstance(first_result.download_backend_info(), BackendInfo)
@@ -432,23 +451,32 @@ def test_get_results_for_incomplete_execute(
         with pytest.raises(qnx_exc.JobError):
             qnx.jobs.wait_for(execute_job_ref)
 
+        with pytest.raises(qnx_exc.ResourceFetchFailed):
+            qnx.jobs.results(execute_job_ref)
+
         incomplete_results = qnx.jobs.results(execute_job_ref, allow_incomplete=True)
 
         # wait for the ZZPhase circuit execution to complete
         for _ in range(10):
             incomplete_results = qnx.jobs.results(
                 execute_job_ref, allow_incomplete=True
             )
-            if len(incomplete_results) > 0:
+            if any(isinstance(r, ExecutionResultRef) for r in incomplete_results):
                 break
             sleep(10)
 
         # we expect the CX circuit to fail on H1-1LE, but the ZZPhase circuit should succeed
-        assert len(incomplete_results) == 1
+        assert len(incomplete_results) == 2
+        first_item, second_item = incomplete_results[0], incomplete_results[1]
+        assert isinstance(first_item, IncompleteJobItemRef)
+        assert first_item.id == UUID(int=0)
+        assert isinstance(first_item.job_item_integer_id, int)
+        assert first_item.last_status == JobStatusEnum.ERROR
 
-        assert isinstance(incomplete_results[0].get_input(), CircuitRef)
-        assert isinstance(incomplete_results[0].download_result(), BackendResult)
-        assert isinstance(incomplete_results[0].download_backend_info(), BackendInfo)
+        assert isinstance(second_item, ExecutionResultRef)
+        assert isinstance(second_item.get_input(), CircuitRef)
+        assert isinstance(second_item.download_result(), BackendResult)
+        assert isinstance(second_item.download_backend_info(), BackendInfo)
 
 
 def test_wait_for_raises_on_job_error(
@@ -523,12 +551,13 @@ def test_results_not_available_error(
         execute_results = qnx.jobs.results(execute_job_ref)
 
         assert len(execute_results) == 1
+        execution_result = execute_results[0]
+        assert isinstance(execution_result, ExecutionResultRef)
+        assert isinstance(execution_result.get_input(), CircuitRef)
 
-        assert isinstance(execute_results[0].get_input(), CircuitRef)
-
-        assert isinstance(execute_results[0].download_result(), BackendResult)
+        assert isinstance(execution_result.download_result(), BackendResult)
 
-        assert isinstance(execute_results[0].download_backend_info(), BackendInfo)
+        assert isinstance(execution_result.download_backend_info(), BackendInfo)
 
 
 def test_submit_under_user_group(

integration/test_qir.py

@@ -13,7 +13,14 @@
 
 import qnexus as qnx
 from qnexus.models.annotations import PropertiesDict
-from qnexus.models.references import ProjectRef, QIRRef, QIRResult, Ref, ResultVersions
+from qnexus.models.references import (
+    ExecutionResultRef,
+    ProjectRef,
+    QIRRef,
+    QIRResult,
+    Ref,
+    ResultVersions,
+)
 
 
 def test_qir_create_and_update(
@@ -156,12 +163,16 @@ def test_execution(
         assert len(results) == 1
         result_ref = results[0]
 
+        assert isinstance(result_ref, ExecutionResultRef)
         assert isinstance(result_ref.download_backend_info(), BackendInfo)
         assert isinstance(result_ref.get_input(), QIRRef)
 
         assert result_ref.get_input().id == qir_program_ref.id
 
-        qir_result = qnx.jobs.results(job_ref)[0].download_result()
+        qir_result_ref = qnx.jobs.results(job_ref)[0]
+
+        assert isinstance(qir_result_ref, ExecutionResultRef)
+        qir_result = qir_result_ref.download_result()
         assert isinstance(qir_result, BackendResult)
         assert qir_result.get_counts() == Counter({(0, 0, 0): 10})
         assert qir_result.get_bitlist() == [Bit("c", 2), Bit("c", 1), Bit("c", 0)]
@@ -193,17 +204,17 @@ def test_execution_on_NG_devices(
 
         qnx.jobs.wait_for(job_ref)
 
-        results = qnx.jobs.results(job_ref)[0].download_result()
+        result_ref = qnx.jobs.results(job_ref)[0]
+        assert isinstance(result_ref, ExecutionResultRef)
+        results = result_ref.download_result()
         # Assert this is a QIR compliant result
         assert isinstance(results, QIRResult)
         escaped_results = results.results.encode("unicode_escape").decode()
         assert "HEADER\\tschema_id\\tlabeled" in escaped_results
         # Can't assert the value is the same, so just check the output is there
         assert "OUTPUT\\tTUPLE\\t2\\tt0" in escaped_results
 
-        v4_results = qnx.jobs.results(job_ref)[0].download_result(
-            version=ResultVersions.RAW
-        )
+        v4_results = result_ref.download_result(version=ResultVersions.RAW)
         # Assert this is in v4 format
         assert isinstance(v4_results, QsysResult)
         assert v4_results.results[0].entries[0][0] == "USER:QIRTUPLE:t0"

integration/test_qsys_jobs.py

@@ -11,7 +11,12 @@
 from quantinuum_schemas.models.backend_config import BasicEmulatorConfig
 
 import qnexus as qnx
-from qnexus.models.references import HUGRRef, ProjectRef, ResultVersions
+from qnexus.models.references import (
+    ExecutionResultRef,
+    HUGRRef,
+    ProjectRef,
+    ResultVersions,
+)
 
 
 def prepare_teleportation() -> Any:
@@ -77,6 +82,7 @@ def test_guppy_execution(
         assert len(results) == 1
         result_ref = results[0]
 
+        assert isinstance(result_ref, ExecutionResultRef)
         assert isinstance(result_ref.download_backend_info(), BackendInfo)
         assert isinstance(result_ref.get_input(), HUGRRef)
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "qnexus"
-version = "0.33.0"
+version = "0.34.0"
 description = "Quantinuum Nexus python client."
 authors = [
     {name = "Vanya Eccles", email = "vanya.eccles@quantinuum.com"},

qnexus/client/jobs/__init__.py

@@ -53,6 +53,7 @@
     ExecutionResult,
     ExecutionResultRef,
     GpuDecoderConfigRef,
+    IncompleteJobItemRef,
     JobRef,
     JobType,
     ProjectRef,
@@ -443,19 +444,22 @@ def _process_exception(exc: Exception) -> Exception | None:
 @overload
 def results(
     job: CompileJobRef, allow_incomplete: bool = False
-) -> DataframableList[CompilationResultRef]: ...
+) -> DataframableList[CompilationResultRef | IncompleteJobItemRef]: ...
 
 
 @overload
 def results(
     job: ExecuteJobRef, allow_incomplete: bool = False
-) -> DataframableList[ExecutionResultRef]: ...
+) -> DataframableList[ExecutionResultRef | IncompleteJobItemRef]: ...
 
 
 def results(
     job: CompileJobRef | ExecuteJobRef,
     allow_incomplete: bool = False,
-) -> DataframableList[CompilationResultRef] | DataframableList[ExecutionResultRef]:
+) -> (
+    DataframableList[CompilationResultRef | IncompleteJobItemRef]
+    | DataframableList[ExecutionResultRef | IncompleteJobItemRef]
+):
     """Get the ResultRefs from a JobRef, if the job is complete.
     To enable fetching results from Jobs with incomplete items, set allow_incomplete=True.
     """
@@ -557,9 +561,18 @@ def compile(
 
     compile_results = results(compile_job_ref)
 
-    return DataframableList(
-        [compile_result.get_output() for compile_result in compile_results]
-    )
+    compiled_circuits: list[CircuitRef] = []
+    for compile_result in compile_results:
+        if isinstance(compile_result, CompilationResultRef):
+            compiled_circuits.append(compile_result.get_output())
+        elif isinstance(compile_result, IncompleteJobItemRef):
+            raise qnx_exc.ResourceFetchFailed(
+                f"Compile job item {compile_result.job_item_integer_id} is in status {compile_result.last_status}"
+            )
+        else:
+            assert_never(compile_result)
+
+    return DataframableList(compiled_circuits)
 
 
 @accept_circuits_for_programs
@@ -607,7 +620,18 @@ def execute(
 
     execute_results = results(execute_job_ref)
 
-    return [result.download_result() for result in execute_results]
+    ex_results: list[ExecutionResult] = []
+    for result in execute_results:
+        if isinstance(result, ExecutionResultRef):
+            ex_results.append(result.download_result())
+        elif isinstance(result, IncompleteJobItemRef):
+            raise qnx_exc.ResourceFetchFailed(
+                f"Compile job item {result.job_item_integer_id} is in status {result.last_status}"
+            )
+        else:
+            assert_never(result)
+
+    return ex_results
 
 
 def cost(job: CompileJobRef | ExecuteJobRef) -> float: