def run_job_over_stream(
        self,
        *,
        project_id: str,
        program_id: str,
        code: any_pb2.Any,
        run_context: any_pb2.Any,
        program_description: str | None = None,
        program_labels: dict[str, str] | None = None,
        job_id: str,
        priority: int | None = None,
        job_description: str | None = None,
        job_labels: dict[str, str] | None = None,
        processor_id: str = "",
        run_name: str = "",
        snapshot_id: str = "",
        device_config_name: str = "",
    ) -> duet.AwaitableFuture[quantum.QuantumResult | quantum.QuantumJob]:
        """Runs a job with the given program and job information over a stream.

        Sends the request over the Quantum Engine QuantumRunStream bidirectional stream, and returns
        a future for the stream response. The future will be completed with a `QuantumResult` if
        the job is successful; otherwise, it will be completed with a QuantumJob.

        Args:
            project_id: A project_id of the parent Google Cloud Project.
            program_id: Unique ID of the program within the parent project.
            code: Properly serialized program code.
            run_context: Properly serialized run context.
            program_description: An optional description to set on the program.
            program_labels: Optional set of labels to set on the program.
            job_id: Unique ID of the job within the parent program.
            priority: Optional priority to run at, 0-1000.
            job_description: Optional description to set on the job.
            job_labels: Optional set of labels to set on the job.
            processor_id: Processor id for running the program.
            run_name: A unique identifier representing an automation run for the
                specified processor. An Automation Run contains a collection of
                device configurations for a processor. If specified, `processor_id`
                is required to be set.
            snapshot_id: A unique identifier for an immutable snapshot reference.
                A snapshot contains a collection of device configurations for the
                processor.
            device_config_name: An identifier used to select the processor configuration
                utilized to run the job. A configuration identifies the set of
                available qubits, couplers, and supported gates in the processor.
                If specified, `processor_id` is required to be set.

        Returns:
            A future for the job result, or the job if the job has failed.

        Raises:
            ValueError: If the priority is not between 0 and 1000.
            ValueError: If `processor_id` is not set.
            ValueError: If only one of `run_name` and `device_config_name` are specified.
            ValueError: If both `run_name` and `snapshot_id` are specified.
        """
        # Check program to run and program parameters.
        if priority and not 0 <= priority < 1000:
            raise ValueError('priority must be between 0 and 1000')
        if not processor_id:
            raise ValueError('Must specify a processor id when creating a job.')
        if run_name and snapshot_id:
            raise ValueError('Cannot specify both `run_name` and `snapshot_id`')
        if (bool(run_name) or bool(snapshot_id)) ^ bool(device_config_name):
            raise ValueError(
                'Cannot specify only one of top level identifier and `device_config_name`'
            )

        project_name = _project_name(project_id)

        program_name = _program_name_from_ids(project_id, program_id)
        program = quantum.QuantumProgram(name=program_name, code=code)
        if program_description:
            program.description = program_description
        if program_labels:
            program.labels.update(program_labels)

        if snapshot_id:
            selector = quantum.DeviceConfigSelector(
                snapshot_id=snapshot_id or None, config_alias=device_config_name
            )
        else:
            selector = quantum.DeviceConfigSelector(
                run_name=run_name or None, config_alias=device_config_name
            )

        job = quantum.QuantumJob(
            name=_job_name_from_ids(project_id, program_id, job_id),
            scheduling_config=quantum.SchedulingConfig(
                processor_selector=quantum.SchedulingConfig.ProcessorSelector(
                    processor=_processor_name_from_ids(project_id, processor_id),
                    device_config_selector=selector,
                )
            ),
            run_context=run_context,
        )
        if priority:
            job.scheduling_config.priority = priority
        if job_description:
            job.description = job_description
        if job_labels:
            job.labels.update(job_labels)

        return self._stream_manager.submit(project_name, program, job)
Runs a job with the given program and job information over a stream.

        Sends the request over the Quantum Engine QuantumRunStream bidirectional stream, and returns
        a future for the stream response. The future will be completed with a `QuantumResult` if
        the job is successful; otherwise, it will be completed with a QuantumJob.

        Args:
            project_id: A project_id of the parent Google Cloud Project.
            program_id: Unique ID of the program within the parent project.
            code: Properly serialized program code.
            run_context: Properly serialized run context.
            program_description: An optional description to set on the program.
            program_labels: Optional set of labels to set on the program.
            job_id: Unique ID of the job within the parent program.
            priority: Optional priority to run at, 0-1000.
            job_description: Optional description to set on the job.
            job_labels: Optional set of labels to set on the job.
            processor_id: Processor id for running the program.
            run_name: A unique identifier representing an automation run for the
                specified processor. An Automation Run contains a collection of
                device configurations for a processor. If specified, `processor_id`
                is required to be set.
            snapshot_id: A unique identifier for an immutable snapshot reference.
                A snapshot contains a collection of device configurations for the
                processor.
            device_config_name: An identifier used to select the processor configuration
                utilized to run the job. A configuration identifies the set of
                available qubits, couplers, and supported gates in the processor.
                If specified, `processor_id` is required to be set.

        Returns:
            A future for the job result, or the job if the job has failed.

        Raises:
            ValueError: If the priority is not between 0 and 1000.
            ValueError: If `processor_id` is not set.
            ValueError: If only one of `run_name` and `device_config_name` are specified.
            ValueError: If both `run_name` and `snapshot_id` are specified.
        
cirq-google/cirq_google/engine/engine_client.py