Make program.run() return an EngineJob so it's consistent with program.run_sweep() (#8048)

Internal request by @eliottrosenberg to have program.run() return an
EngineJob, rather than having the function wait for the EngineJob to
complete before returning.

This interface makes it consistent with existing program.run_sweep()

Fixes b/495930398

Author: mattwenner

cirq-google/cirq_google/engine/engine_program.py

@@ -147,7 +147,7 @@ async def run_async(
         repetitions: int = 1,
         description: str | None = None,
         labels: dict[str, str] | None = None,
-    ) -> cirq.Result:
+    ) -> engine_job.EngineJob:
         """Runs the supplied Circuit via Quantum Engine.
 
         Args:
@@ -171,15 +171,15 @@ async def run_async(
             labels: Optional set of labels to set on the job.
 
         Returns:
-            A single Result for this run.
+            An EngineJob. If this is iterated over it yields one EngineResult.
 
         Raises:
             ValueError: If a processor id hasn't been specified to run the job
             ValueError: If  only one of `run_name` and `device_config_name` are specified.
             ValueError: If either `run_name` and `device_config_name` are set but
                 `processor_id` is empty.
         """
-        job = await self.run_sweep_async(
+        return await self.run_sweep_async(
             job_id=job_id,
             params=[param_resolver],
             repetitions=repetitions,
@@ -190,8 +190,6 @@ async def run_async(
             snapshot_id=snapshot_id,
             device_config_name=device_config_name,
         )
-        results = await job.results_async()
-        return results[0]
 
     run = duet.sync(run_async)
 

cirq-google/cirq_google/engine/engine_program_test.py

@@ -139,7 +139,7 @@ def test_run_delegation(create_job_async, get_results_async):
 
     program = cg.EngineProgram('a', 'b', EngineContext())
     param_resolver = cirq.ParamResolver({})
-    results = program.run(
+    job = program.run(
         job_id='steve',
         repetitions=10,
         param_resolver=param_resolver,
@@ -148,7 +148,9 @@ def test_run_delegation(create_job_async, get_results_async):
         device_config_name="config",
     )
 
-    assert results == cg.EngineResult(
+    (result,) = job
+
+    assert result == cg.EngineResult(
         params=cirq.ParamResolver({'a': 1.0}),
         measurements={'q': np.array([[False], [True], [True], [False]], dtype=bool)},
         job_id='steve',

cirq-google/cirq_google/engine/engine_test.py

@@ -612,7 +612,7 @@ def test_run_multiple_times(client):
         param_resolver=cirq.ParamResolver({'a': 1}),
         run_name="run",
         device_config_name="config_alias",
-    )
+    ).results()
     run_context = v2.run_context_pb2.RunContext()
     client().create_job_async.call_args[1]['run_context'].Unpack(run_context)
     sweeps1 = run_context.parameter_sweeps