async def _manage_execution(
        self,
        request_queue: asyncio.Queue[quantum.QuantumRunStreamRequest | None],
        project_name: str,
        program: quantum.QuantumProgram,
        job: quantum.QuantumJob,
    ) -> quantum.QuantumResult | quantum.QuantumJob:
        """The execution coroutine, an asyncio coroutine to manage the lifecycle of a job execution.

        This coroutine sends QuantumRunStream requests to the request iterator and receives
        responses from the ResponseDemux.

        It initially sends a CreateQuantumProgramAndJobRequest, and retries if there is a retryable
        error by sending another request. The exact request type depends on the error.

        There is one execution coroutine per running job submission.

        Args:
            request_queue: The queue used to send requests to the stream coroutine.
            project_name: The full project ID resource path associated with the job.
            program: The Quantum Engine program representing the circuit to be executed.
            job: The Quantum Engine job to be executed.

        Raises:
            concurrent.futures.CancelledError: if either the request is cancelled or the stream
                coroutine is cancelled.
            google.api_core.exceptions.GoogleAPICallError: if the stream breaks with a non-retryable
                error.
            ValueError: if the response is of a type which is not recognized by this client.
        """
        create_program_and_job_request = quantum.QuantumRunStreamRequest(
            parent=project_name,
            create_quantum_program_and_job=quantum.CreateQuantumProgramAndJobRequest(
                parent=project_name, quantum_program=program, quantum_job=job
            ),
        )

        current_request = create_program_and_job_request
        while True:
            try:
                current_request.message_id = self._generate_message_id()
                response_future = self._response_demux.subscribe(current_request.message_id)
                await request_queue.put(current_request)
                response = await response_future

            # Broken stream
            except google_exceptions.GoogleAPICallError as e:
                if not _is_retryable_error(e):
                    raise e

                # Retry
                current_request = _to_get_result_request(create_program_and_job_request)
                continue
                # TODO(#5996) add exponential backoff

            # Either when this request is canceled or the _manage_stream() loop is canceled.
            except asyncio.CancelledError:
                # TODO(#5996) Consider moving the request future cancellation logic into a future
                # done callback, so that the the cancellation caller can wait for it to complete.
                # TODO(#5996) Check the condition that response_future is not done before
                # cancelling, once request cancellation is moved to a callback.
                if response_future is not None:
                    response_future.cancel()
                    await self._cancel(job.name)
                raise

            # Response handling
            if 'result' in response:
                return response.result
            elif 'job' in response:
                return response.job
            elif 'error' in response:
                current_request = _get_retry_request_or_raise(
                    response.error,
                    current_request,
                    create_program_and_job_request,
                    _to_create_job_request(create_program_and_job_request),
                    _to_get_result_request(create_program_and_job_request),
                )
                continue
            else:  # pragma: no cover
                raise ValueError(
                    'The Quantum Engine response type is not recognized by this client. '
                    'This may be due to an outdated version of cirq-google'
                )
The execution coroutine, an asyncio coroutine to manage the lifecycle of a job execution.

        This coroutine sends QuantumRunStream requests to the request iterator and receives
        responses from the ResponseDemux.

        It initially sends a CreateQuantumProgramAndJobRequest, and retries if there is a retryable
        error by sending another request. The exact request type depends on the error.

        There is one execution coroutine per running job submission.

        Args:
            request_queue: The queue used to send requests to the stream coroutine.
            project_name: The full project ID resource path associated with the job.
            program: The Quantum Engine program representing the circuit to be executed.
            job: The Quantum Engine job to be executed.

        Raises:
            concurrent.futures.CancelledError: if either the request is cancelled or the stream
                coroutine is cancelled.
            google.api_core.exceptions.GoogleAPICallError: if the stream breaks with a non-retryable
                error.
            ValueError: if the response is of a type which is not recognized by this client.
        
cirq-google/cirq_google/engine/stream_manager.py