From d2d1b3b71551f3f38927301ebf153f9368048f72 Mon Sep 17 00:00:00 2001
From: dyates <devyates@google.com>
Date: Fri, 19 Sep 2025 23:18:08 +0000
Subject: [PATCH 01/12] Add new `get_sampler*` functions. Refactor existing
 `get_sampler`.

---
 cirq-google/cirq_google/engine/engine.py      | 75 +++++++++++++---
 .../cirq_google/engine/engine_processor.py    | 85 ++++++++++++++-----
 .../engine/engine_processor_test.py           | 76 ++++++++---------
 cirq-google/cirq_google/engine/engine_test.py |  2 +-
 4 files changed, 165 insertions(+), 73 deletions(-)

diff --git a/cirq-google/cirq_google/engine/engine.py b/cirq-google/cirq_google/engine/engine.py
index fd9e614796d..6594c18de92 100644
--- a/cirq-google/cirq_google/engine/engine.py
+++ b/cirq-google/cirq_google/engine/engine.py
@@ -589,26 +589,82 @@ def get_processor(self, processor_id: str) -> engine_processor.EngineProcessor:
         """
         return engine_processor.EngineProcessor(self.project_id, processor_id, self.context)
 
+
+    def get_sampler_from_run_name(
+        self,
+        processor_id: str,
+        device_config_name: str,
+        run_name: str = 'default',
+        max_concurrent_jobs: int = 10,
+    ) -> cirq_google.ProcessorSampler:
+        """Returns a sampler backed by the engine and given `run_name`.
+
+        Args:
+            run_name: A unique identifier representing an automation run for the
+                processor. An Automation Run 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.
+            max_concurrent_jobs: The maximum number of jobs to be sent
+                simultaneously to the Engine. This client-side throttle can be
+                used to proactively reduce load to the backends and avoid quota
+                violations when pipelining circuit executions.
+
+        Returns:
+            A `cirq.Sampler` instance (specifically a `engine_sampler.ProcessorSampler`
+            that will send circuits to the Quantum Computing Service
+            when sampled.
+        """
+        self.get_processor(processor_id).get_sampler_from_run_name(
+            run_name=run_name,
+            device_config_name=device_config_name,
+            max_concurrent_jobs=max_concurrent_jobs
+        )
+
+    def get_sampler_from_snapshot_id(
+        self,
+        processor_id: str,
+        snapshot_id: str,
+        device_config_name: str,
+        max_concurrent_jobs: int = 10,
+    ) -> cirq_google.ProcessorSampler:
+        """Returns a sampler backed by the engine.
+        Args:
+            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.
+            snapshot_id: A unique identifier for an immutable snapshot reference.
+                A snapshot contains a collection of device configurations for the
+                processor.
+            max_concurrent_jobs: The maximum number of jobs to be sent
+                simultaneously to the Engine. This client-side throttle can be
+                used to proactively reduce load to the backends and avoid quota
+                violations when pipelining circuit executions.
+
+        Returns:
+            A `cirq.Sampler` instance (specifically a `engine_sampler.ProcessorSampler`
+            that will send circuits to the Quantum Computing Service
+            when sampled.
+        """
+        return self.get_processor(processor_id).get_sampler_from_snapshot_id(
+            snapshot_id=snapshot_id,
+            device_config_name=device_config_name,
+            max_concurrent_jobs=10
+        )
+
     def get_sampler(
         self,
         processor_id: str | list[str],
-        run_name: str = "",
-        device_config_name: str = "",
-        snapshot_id: str = "",
         max_concurrent_jobs: int = 10,
     ) -> cirq_google.ProcessorSampler:
         """Returns a sampler backed by the engine.
 
         Args:
             processor_id: String identifier of which processor should be used to sample.
-            run_name: A unique identifier representing an automation run for the
-                processor. An Automation Run 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.
-            snapshot_id: A unique identifier for an immutable snapshot reference. A
-                snapshot contains a collection of device configurations for the processor.
             max_concurrent_jobs: The maximum number of jobs to be sent
                 concurrently to the Engine. This client-side throttle can be
                 used to proactively reduce load to the backends and avoid quota
@@ -629,9 +685,6 @@ def get_sampler(
                 'you need to specify a list.'
             )
         return self.get_processor(processor_id).get_sampler(
-            run_name=run_name,
-            device_config_name=device_config_name,
-            snapshot_id=snapshot_id,
             max_concurrent_jobs=max_concurrent_jobs,
         )
 
diff --git a/cirq-google/cirq_google/engine/engine_processor.py b/cirq-google/cirq_google/engine/engine_processor.py
index 2f4c3dbde2d..fe29fc1070c 100644
--- a/cirq-google/cirq_google/engine/engine_processor.py
+++ b/cirq-google/cirq_google/engine/engine_processor.py
@@ -93,18 +93,46 @@ def engine(self) -> engine_base.Engine:
 
         return engine_base.Engine(self.project_id, context=self.context)
 
-    def get_sampler(
+    def get_sampler_from_run_name(
         self,
-        run_name: str = "",
-        device_config_name: str = "",
-        snapshot_id: str = "",
+        device_config_name: str,
+        run_name: str = 'default',
         max_concurrent_jobs: int = 10,
     ) -> cg.engine.ProcessorSampler:
-        """Returns a sampler backed by the engine.
+        """Returns a sampler backed by the engine and given `run_name`.
+
         Args:
             run_name: A unique identifier representing an automation run for the
                 processor. An Automation Run 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.
+            max_concurrent_jobs: The maximum number of jobs to be sent
+                simultaneously to the Engine. This client-side throttle can be
+                used to proactively reduce load to the backends and avoid quota
+                violations when pipelining circuit executions.
+
+        Returns:
+            A `cirq.Sampler` instance (specifically a `engine_sampler.ProcessorSampler`
+            that will send circuits to the Quantum Computing Service
+            when sampled.
+        """
+        return processor_sampler.ProcessorSampler(
+            processor=self,
+            run_name=run_name,
+            device_config_name=device_config_name,
+            max_concurrent_jobs=max_concurrent_jobs,
+        )
+
+    def get_sampler_from_snapshot_id(
+        self,
+        snapshot_id: str,
+        device_config_name: str,
+        max_concurrent_jobs: int = 10,
+    ) -> cg.engine.ProcessorSampler:
+        """Returns a sampler backed by the engine.
+        Args:
             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.
@@ -120,29 +148,42 @@ def get_sampler(
             A `cirq.Sampler` instance (specifically a `engine_sampler.ProcessorSampler`
             that will send circuits to the Quantum Computing Service
             when sampled.
+        """
+        return processor_sampler.ProcessorSampler(
+            processor=self,
+            snapshot_id=snapshot_id,
+            device_config_name=device_config_name,
+            max_concurrent_jobs=max_concurrent_jobs,
+        )
 
-        Raises:
-            ValueError: If only one of `run_name` and `device_config_name` are specified.
-            ValueError: If both `run_name` and `snapshot_id` are specified.
+    def get_sampler(
+        self,
+        device_config_name: str | None = None,
+        max_concurrent_jobs: int = 10,
+    ) -> cg.engine.ProcessorSampler:
+        """Returns the default sampler backed by the engine.
+
+        Args:
+            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.
+            max_concurrent_jobs: The maximum number of jobs to be sent
+                simultaneously to the Engine. This client-side throttle can be
+                used to proactively reduce load to the backends and avoid quota
+                violations when pipelining circuit executions.
+
+        Returns:
+            A `cirq.Sampler` instance (specifically a `engine_sampler.ProcessorSampler`
+            that will send circuits to the Quantum Computing Service
+            when sampled.
 
         """
         processor = self._inner_processor()
-        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`'
-            )
-        # If not provided, initialize the sampler with the Processor's default values.
-        if not run_name and not device_config_name and not snapshot_id:
-            run_name = processor.default_device_config_key.run
-            device_config_name = processor.default_device_config_key.config_alias
-            snapshot_id = processor.default_device_config_key.snapshot_id
+        
         return processor_sampler.ProcessorSampler(
             processor=self,
-            run_name=run_name,
-            snapshot_id=snapshot_id,
-            device_config_name=device_config_name,
+            run_name=processor.default_device_config_key.run,
+            device_config_name=device_config_name if device_config_name else processor.default_device_config_key.config_alias,
             max_concurrent_jobs=max_concurrent_jobs,
         )
 
diff --git a/cirq-google/cirq_google/engine/engine_processor_test.py b/cirq-google/cirq_google/engine/engine_processor_test.py
index dee637b5e25..339cb68eaf0 100644
--- a/cirq-google/cirq_google/engine/engine_processor_test.py
+++ b/cirq-google/cirq_google/engine/engine_processor_test.py
@@ -310,72 +310,70 @@ def test_get_missing_device():
     with pytest.raises(ValueError, match='device specification'):
         _ = processor.get_device()
 
+def test_get_sampler_from_run_name() -> None:
+    processor = cg.EngineProcessor(
+        'a',
+        'p',
+        EngineContext(),
+    )
+    run_name = 'test_run_name'
+    device_config_name = 'test_device_name'
 
-def test_get_sampler_initializes_default_device_configuration() -> None:
+    sampler = processor.get_sampler_from_run_name(
+        run_name=run_name,
+        device_config_name=device_config_name
+    )
+
+    assert sampler.run_name == run_name
+    assert sampler.device_config_name == device_config_name
+
+def test_get_sampler_from_run_name_with_default_run_name() -> None:
     processor = cg.EngineProcessor(
         'a',
         'p',
         EngineContext(),
-        _processor=quantum.QuantumProcessor(
-            default_device_config_key=quantum.DeviceConfigKey(
-                run="run", config_alias="config_alias"
-            )
-        ),
     )
-    sampler = processor.get_sampler()
+    device_config_name = 'test_device_name'
 
-    assert sampler.run_name == "run"
-    assert sampler.device_config_name == "config_alias"
+    sampler = processor.get_sampler_from_run_name(
+        device_config_name=device_config_name
+    )
 
+    assert sampler.run_name == 'default'
+    assert sampler.device_config_name == device_config_name
 
-def test_get_sampler_uses_custom_default_device_configuration_key() -> None:
+def test_get_sampler_from_snapshot_id() -> None:
     processor = cg.EngineProcessor(
         'a',
         'p',
         EngineContext(),
-        _processor=quantum.QuantumProcessor(
-            default_device_config_key=quantum.DeviceConfigKey(
-                run="default_run", config_alias="default_config_alias"
-            )
-        ),
     )
-    sampler = processor.get_sampler(run_name="run1", device_config_name="config_alias1")
+    snapshot_id = 'test_snapshot'
+    device_config_name = 'test_device_name'
 
-    assert sampler.run_name == "run1"
-    assert sampler.device_config_name == "config_alias1"
+    sampler = processor.get_sampler_from_snapshot_id(
+        snapshot_id=snapshot_id,
+        device_config_name=device_config_name
+    )
 
+    assert sampler.snapshot_id == snapshot_id
+    assert sampler.device_config_name == device_config_name
 
-@pytest.mark.parametrize(
-    'run, snapshot_id, config_alias, error_message',
-    [
-        ('run', '', '', 'Cannot specify only one of top level identifier and `device_config_name`'),
-        (
-            '',
-            '',
-            'config',
-            'Cannot specify only one of top level identifier and `device_config_name`',
-        ),
-        ('run', 'snapshot_id', 'config', 'Cannot specify both `run_name` and `snapshot_id`'),
-    ],
-)
-def test_get_sampler_with_incomplete_device_configuration_errors(
-    run, snapshot_id, config_alias, error_message
-) -> None:
+def test_get_sampler_initializes_default_device_configuration() -> None:
     processor = cg.EngineProcessor(
         'a',
         'p',
         EngineContext(),
         _processor=quantum.QuantumProcessor(
             default_device_config_key=quantum.DeviceConfigKey(
-                run="default_run", config_alias="default_config_alias"
+                run="run", config_alias="config_alias"
             )
         ),
     )
+    sampler = processor.get_sampler()
 
-    with pytest.raises(ValueError, match=error_message):
-        processor.get_sampler(
-            run_name=run, device_config_name=config_alias, snapshot_id=snapshot_id
-        )
+    assert sampler.run_name == "run"
+    assert sampler.device_config_name == "config_alias"
 
 
 @mock.patch('cirq_google.engine.engine_client.EngineClient.get_processor_async')
diff --git a/cirq-google/cirq_google/engine/engine_test.py b/cirq-google/cirq_google/engine/engine_test.py
index 86ea480f6f6..ca88763fda6 100644
--- a/cirq-google/cirq_google/engine/engine_test.py
+++ b/cirq-google/cirq_google/engine/engine_test.py
@@ -296,7 +296,7 @@ def test_engine_get_sampler_with_snapshot_id_passes_to_unary_rpc(client):
         project_id='proj',
         context=EngineContext(service_args={'client_info': 1}, enable_streaming=False),
     )
-    sampler = engine.get_sampler('mysim', device_config_name="config", snapshot_id="123")
+    sampler = engine.get_sampler_from_snapshot_id('mysim', device_config_name="config", snapshot_id="123")
     _ = sampler.run_sweep(_CIRCUIT, params=[cirq.ParamResolver({'a': 1})])
 
     kwargs = client().create_job_async.call_args_list[0].kwargs

From 396438a369e901466afdd0c481a71a80f72611f8 Mon Sep 17 00:00:00 2001
From: dyates <devyates@google.com>
Date: Thu, 25 Sep 2025 00:35:51 +0000
Subject: [PATCH 02/12] Change default concurrent jobs to 100.

---
 cirq-google/cirq_google/engine/engine.py           | 4 ++--
 cirq-google/cirq_google/engine/engine_processor.py | 4 ++--
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/cirq-google/cirq_google/engine/engine.py b/cirq-google/cirq_google/engine/engine.py
index 9b2889aa4bd..f0e42bd05b9 100644
--- a/cirq-google/cirq_google/engine/engine.py
+++ b/cirq-google/cirq_google/engine/engine.py
@@ -596,7 +596,7 @@ def get_sampler_from_run_name(
         processor_id: str,
         device_config_name: str,
         run_name: str = 'default',
-        max_concurrent_jobs: int = 10,
+        max_concurrent_jobs: int = 100,
     ) -> cirq_google.ProcessorSampler:
         """Returns a sampler backed by the engine and given `run_name`.
 
@@ -628,7 +628,7 @@ def get_sampler_from_snapshot_id(
         processor_id: str,
         snapshot_id: str,
         device_config_name: str,
-        max_concurrent_jobs: int = 10,
+        max_concurrent_jobs: int = 100,
     ) -> cirq_google.ProcessorSampler:
         """Returns a sampler backed by the engine.
         Args:
diff --git a/cirq-google/cirq_google/engine/engine_processor.py b/cirq-google/cirq_google/engine/engine_processor.py
index 62ec458bfad..5cae60208be 100644
--- a/cirq-google/cirq_google/engine/engine_processor.py
+++ b/cirq-google/cirq_google/engine/engine_processor.py
@@ -136,7 +136,7 @@ def get_sampler_from_snapshot_id(
         self,
         snapshot_id: str,
         device_config_name: str,
-        max_concurrent_jobs: int = 10,
+        max_concurrent_jobs: int = 100,
     ) -> cg.engine.ProcessorSampler:
         """Returns a sampler backed by the engine.
         Args:
@@ -166,7 +166,7 @@ def get_sampler_from_snapshot_id(
     def get_sampler(
         self,
         device_config_name: str | None = None,
-        max_concurrent_jobs: int = 10,
+        max_concurrent_jobs: int = 100,
     ) -> cg.engine.ProcessorSampler:
         """Returns the default sampler backed by the engine.
 

From fcf006d3f4b0c244b46ecb7c3cf5819896597b95 Mon Sep 17 00:00:00 2001
From: dyates <devyates@google.com>
Date: Thu, 25 Sep 2025 00:37:32 +0000
Subject: [PATCH 03/12] Format checks.

---
 cirq-google/cirq_google/engine/engine.py      | 11 ++-----
 .../cirq_google/engine/engine_processor.py    | 17 +++++-----
 .../engine/engine_processor_test.py           | 32 ++++++-------------
 cirq-google/cirq_google/engine/engine_test.py |  4 ++-
 4 files changed, 24 insertions(+), 40 deletions(-)

diff --git a/cirq-google/cirq_google/engine/engine.py b/cirq-google/cirq_google/engine/engine.py
index f0e42bd05b9..ee4ef954614 100644
--- a/cirq-google/cirq_google/engine/engine.py
+++ b/cirq-google/cirq_google/engine/engine.py
@@ -590,7 +590,6 @@ def get_processor(self, processor_id: str) -> engine_processor.EngineProcessor:
         """
         return engine_processor.EngineProcessor(self.project_id, processor_id, self.context)
 
-
     def get_sampler_from_run_name(
         self,
         processor_id: str,
@@ -620,7 +619,7 @@ def get_sampler_from_run_name(
         self.get_processor(processor_id).get_sampler_from_run_name(
             run_name=run_name,
             device_config_name=device_config_name,
-            max_concurrent_jobs=max_concurrent_jobs
+            max_concurrent_jobs=max_concurrent_jobs,
         )
 
     def get_sampler_from_snapshot_id(
@@ -649,9 +648,7 @@ def get_sampler_from_snapshot_id(
             when sampled.
         """
         return self.get_processor(processor_id).get_sampler_from_snapshot_id(
-            snapshot_id=snapshot_id,
-            device_config_name=device_config_name,
-            max_concurrent_jobs=10
+            snapshot_id=snapshot_id, device_config_name=device_config_name, max_concurrent_jobs=10
         )
 
     def get_sampler(
@@ -688,9 +685,7 @@ def get_sampler(
                 'to get_sampler() no longer supported. Use Engine.run() instead if '
                 'you need to specify a list.'
             )
-        return self.get_processor(processor_id).get_sampler(
-            max_concurrent_jobs=max_concurrent_jobs,
-        )
+        return self.get_processor(processor_id).get_sampler(max_concurrent_jobs=max_concurrent_jobs)
 
     async def get_processor_config_from_snapshot_async(
         self, processor_id: str, snapshot_id: str, config_name: str = 'default'
diff --git a/cirq-google/cirq_google/engine/engine_processor.py b/cirq-google/cirq_google/engine/engine_processor.py
index 5cae60208be..9a881620c2a 100644
--- a/cirq-google/cirq_google/engine/engine_processor.py
+++ b/cirq-google/cirq_google/engine/engine_processor.py
@@ -133,10 +133,7 @@ def get_sampler_from_run_name(
         )
 
     def get_sampler_from_snapshot_id(
-        self,
-        snapshot_id: str,
-        device_config_name: str,
-        max_concurrent_jobs: int = 100,
+        self, snapshot_id: str, device_config_name: str, max_concurrent_jobs: int = 100
     ) -> cg.engine.ProcessorSampler:
         """Returns a sampler backed by the engine.
         Args:
@@ -164,9 +161,7 @@ def get_sampler_from_snapshot_id(
         )
 
     def get_sampler(
-        self,
-        device_config_name: str | None = None,
-        max_concurrent_jobs: int = 100,
+        self, device_config_name: str | None = None, max_concurrent_jobs: int = 100
     ) -> cg.engine.ProcessorSampler:
         """Returns the default sampler backed by the engine.
 
@@ -186,11 +181,15 @@ def get_sampler(
 
         """
         processor = self._inner_processor()
-        
+
         return processor_sampler.ProcessorSampler(
             processor=self,
             run_name=processor.default_device_config_key.run,
-            device_config_name=device_config_name if device_config_name else processor.default_device_config_key.config_alias,
+            device_config_name=(
+                device_config_name
+                if device_config_name
+                else processor.default_device_config_key.config_alias
+            ),
             max_concurrent_jobs=max_concurrent_jobs,
         )
 
diff --git a/cirq-google/cirq_google/engine/engine_processor_test.py b/cirq-google/cirq_google/engine/engine_processor_test.py
index b2c98741be5..85353b96036 100644
--- a/cirq-google/cirq_google/engine/engine_processor_test.py
+++ b/cirq-google/cirq_google/engine/engine_processor_test.py
@@ -323,55 +323,43 @@ def test_get_missing_device():
     with pytest.raises(ValueError, match='device specification'):
         _ = processor.get_device()
 
+
 def test_get_sampler_from_run_name() -> None:
-    processor = cg.EngineProcessor(
-        'a',
-        'p',
-        EngineContext(),
-    )
+    processor = cg.EngineProcessor('a', 'p', EngineContext())
     run_name = 'test_run_name'
     device_config_name = 'test_device_name'
 
     sampler = processor.get_sampler_from_run_name(
-        run_name=run_name,
-        device_config_name=device_config_name
+        run_name=run_name, device_config_name=device_config_name
     )
 
     assert sampler.run_name == run_name
     assert sampler.device_config_name == device_config_name
 
+
 def test_get_sampler_from_run_name_with_default_run_name() -> None:
-    processor = cg.EngineProcessor(
-        'a',
-        'p',
-        EngineContext(),
-    )
+    processor = cg.EngineProcessor('a', 'p', EngineContext())
     device_config_name = 'test_device_name'
 
-    sampler = processor.get_sampler_from_run_name(
-        device_config_name=device_config_name
-    )
+    sampler = processor.get_sampler_from_run_name(device_config_name=device_config_name)
 
     assert sampler.run_name == 'default'
     assert sampler.device_config_name == device_config_name
 
+
 def test_get_sampler_from_snapshot_id() -> None:
-    processor = cg.EngineProcessor(
-        'a',
-        'p',
-        EngineContext(),
-    )
+    processor = cg.EngineProcessor('a', 'p', EngineContext())
     snapshot_id = 'test_snapshot'
     device_config_name = 'test_device_name'
 
     sampler = processor.get_sampler_from_snapshot_id(
-        snapshot_id=snapshot_id,
-        device_config_name=device_config_name
+        snapshot_id=snapshot_id, device_config_name=device_config_name
     )
 
     assert sampler.snapshot_id == snapshot_id
     assert sampler.device_config_name == device_config_name
 
+
 def test_get_sampler_initializes_default_device_configuration() -> None:
     processor = cg.EngineProcessor(
         'a',
diff --git a/cirq-google/cirq_google/engine/engine_test.py b/cirq-google/cirq_google/engine/engine_test.py
index 67814422c6d..fcb77ff84e1 100644
--- a/cirq-google/cirq_google/engine/engine_test.py
+++ b/cirq-google/cirq_google/engine/engine_test.py
@@ -296,7 +296,9 @@ def test_engine_get_sampler_with_snapshot_id_passes_to_unary_rpc(client):
         project_id='proj',
         context=EngineContext(service_args={'client_info': 1}, enable_streaming=False),
     )
-    sampler = engine.get_sampler_from_snapshot_id('mysim', device_config_name="config", snapshot_id="123")
+    sampler = engine.get_sampler_from_snapshot_id(
+        'mysim', device_config_name="config", snapshot_id="123"
+    )
     _ = sampler.run_sweep(_CIRCUIT, params=[cirq.ParamResolver({'a': 1})])
 
     kwargs = client().create_job_async.call_args_list[0].kwargs

From 36fc45da34b4d12819afb63031f0049ad9b87104 Mon Sep 17 00:00:00 2001
From: dyates <devyates@google.com>
Date: Thu, 25 Sep 2025 23:19:42 +0000
Subject: [PATCH 04/12] Merge fixes.

---
 cirq-google/cirq_google/engine/engine.py      |  7 +---
 .../cirq_google/engine/engine_processor.py    | 25 ++++++++++----
 .../engine/engine_processor_test.py           | 33 +++++++++++++++++--
 3 files changed, 50 insertions(+), 15 deletions(-)

diff --git a/cirq-google/cirq_google/engine/engine.py b/cirq-google/cirq_google/engine/engine.py
index ee4ef954614..df9e59526b9 100644
--- a/cirq-google/cirq_google/engine/engine.py
+++ b/cirq-google/cirq_google/engine/engine.py
@@ -652,12 +652,7 @@ def get_sampler_from_snapshot_id(
         )
 
     def get_sampler(
-        self,
-        processor_id: str | list[str],
-        run_name: str = "",
-        device_config_name: str = "",
-        snapshot_id: str = "",
-        max_concurrent_jobs: int = 100,
+        self, processor_id: str | list[str], max_concurrent_jobs: int = 100
     ) -> cirq_google.ProcessorSampler:
         """Returns a sampler backed by the engine.
 
diff --git a/cirq-google/cirq_google/engine/engine_processor.py b/cirq-google/cirq_google/engine/engine_processor.py
index 9a881620c2a..e955a235cdd 100644
--- a/cirq-google/cirq_google/engine/engine_processor.py
+++ b/cirq-google/cirq_google/engine/engine_processor.py
@@ -101,9 +101,8 @@ def engine(self) -> engine_base.Engine:
 
     def get_sampler_from_run_name(
         self,
-        run_name: str = "",
-        device_config_name: str = "",
-        snapshot_id: str = "",
+        run_name: str = 'default',
+        device_config_name: str | None = None,
         max_concurrent_jobs: int = 100,
     ) -> cg.engine.ProcessorSampler:
         """Returns a sampler backed by the engine and given `run_name`.
@@ -125,15 +124,23 @@ def get_sampler_from_run_name(
             that will send circuits to the Quantum Computing Service
             when sampled.
         """
+        processor = self._inner_processor()
         return processor_sampler.ProcessorSampler(
             processor=self,
             run_name=run_name,
-            device_config_name=device_config_name,
+            device_config_name=(
+                device_config_name
+                if device_config_name
+                else processor.default_device_config_key.config_alias
+            ),
             max_concurrent_jobs=max_concurrent_jobs,
         )
 
     def get_sampler_from_snapshot_id(
-        self, snapshot_id: str, device_config_name: str, max_concurrent_jobs: int = 100
+        self,
+        snapshot_id: str,
+        device_config_name: str | None = None,
+        max_concurrent_jobs: int = 100,
     ) -> cg.engine.ProcessorSampler:
         """Returns a sampler backed by the engine.
         Args:
@@ -153,10 +160,15 @@ def get_sampler_from_snapshot_id(
             that will send circuits to the Quantum Computing Service
             when sampled.
         """
+        processor = self._inner_processor()
         return processor_sampler.ProcessorSampler(
             processor=self,
             snapshot_id=snapshot_id,
-            device_config_name=device_config_name,
+            device_config_name=(
+                device_config_name
+                if device_config_name
+                else processor.default_device_config_key.config_alias
+            ),
             max_concurrent_jobs=max_concurrent_jobs,
         )
 
@@ -185,6 +197,7 @@ def get_sampler(
         return processor_sampler.ProcessorSampler(
             processor=self,
             run_name=processor.default_device_config_key.run,
+            snapshot_id=processor.default_device_config_key.snapshot_id,
             device_config_name=(
                 device_config_name
                 if device_config_name
diff --git a/cirq-google/cirq_google/engine/engine_processor_test.py b/cirq-google/cirq_google/engine/engine_processor_test.py
index 85353b96036..40dde396fc0 100644
--- a/cirq-google/cirq_google/engine/engine_processor_test.py
+++ b/cirq-google/cirq_google/engine/engine_processor_test.py
@@ -325,7 +325,16 @@ def test_get_missing_device():
 
 
 def test_get_sampler_from_run_name() -> None:
-    processor = cg.EngineProcessor('a', 'p', EngineContext())
+    processor = cg.EngineProcessor(
+        'a',
+        'p',
+        EngineContext(),
+        _processor=quantum.QuantumProcessor(
+            default_device_config_key=quantum.DeviceConfigKey(
+                run="run", config_alias="config_alias"
+            )
+        ),
+    )
     run_name = 'test_run_name'
     device_config_name = 'test_device_name'
 
@@ -338,7 +347,16 @@ def test_get_sampler_from_run_name() -> None:
 
 
 def test_get_sampler_from_run_name_with_default_run_name() -> None:
-    processor = cg.EngineProcessor('a', 'p', EngineContext())
+    processor = cg.EngineProcessor(
+        'a',
+        'p',
+        EngineContext(),
+        _processor=quantum.QuantumProcessor(
+            default_device_config_key=quantum.DeviceConfigKey(
+                run="run", config_alias="config_alias"
+            )
+        ),
+    )
     device_config_name = 'test_device_name'
 
     sampler = processor.get_sampler_from_run_name(device_config_name=device_config_name)
@@ -348,7 +366,16 @@ def test_get_sampler_from_run_name_with_default_run_name() -> None:
 
 
 def test_get_sampler_from_snapshot_id() -> None:
-    processor = cg.EngineProcessor('a', 'p', EngineContext())
+    processor = cg.EngineProcessor(
+        'a',
+        'p',
+        EngineContext(),
+        _processor=quantum.QuantumProcessor(
+            default_device_config_key=quantum.DeviceConfigKey(
+                run="run", config_alias="config_alias"
+            )
+        ),
+    )
     snapshot_id = 'test_snapshot'
     device_config_name = 'test_device_name'
 

From 2a7cf4a4035145398dfc099f6eea939696d782a8 Mon Sep 17 00:00:00 2001
From: dyates <devyates@google.com>
Date: Fri, 26 Sep 2025 00:14:25 +0000
Subject: [PATCH 05/12] More format fixes;Add unit tests for coverage.

---
 cirq-google/cirq_google/engine/engine.py      | 18 ++++++---
 .../engine/engine_processor_test.py           | 32 +++++++++++++---
 cirq-google/cirq_google/engine/engine_test.py | 38 +++++++++++++++++++
 3 files changed, 78 insertions(+), 10 deletions(-)

diff --git a/cirq-google/cirq_google/engine/engine.py b/cirq-google/cirq_google/engine/engine.py
index df9e59526b9..29efb786d3b 100644
--- a/cirq-google/cirq_google/engine/engine.py
+++ b/cirq-google/cirq_google/engine/engine.py
@@ -593,7 +593,7 @@ def get_processor(self, processor_id: str) -> engine_processor.EngineProcessor:
     def get_sampler_from_run_name(
         self,
         processor_id: str,
-        device_config_name: str,
+        device_config_name: str | None = None,
         run_name: str = 'default',
         max_concurrent_jobs: int = 100,
     ) -> cirq_google.ProcessorSampler:
@@ -616,7 +616,7 @@ def get_sampler_from_run_name(
             that will send circuits to the Quantum Computing Service
             when sampled.
         """
-        self.get_processor(processor_id).get_sampler_from_run_name(
+        return self.get_processor(processor_id).get_sampler_from_run_name(
             run_name=run_name,
             device_config_name=device_config_name,
             max_concurrent_jobs=max_concurrent_jobs,
@@ -648,11 +648,16 @@ def get_sampler_from_snapshot_id(
             when sampled.
         """
         return self.get_processor(processor_id).get_sampler_from_snapshot_id(
-            snapshot_id=snapshot_id, device_config_name=device_config_name, max_concurrent_jobs=10
+            snapshot_id=snapshot_id,
+            device_config_name=device_config_name,
+            max_concurrent_jobs=max_concurrent_jobs,
         )
 
     def get_sampler(
-        self, processor_id: str | list[str], max_concurrent_jobs: int = 100
+        self,
+        processor_id: str | list[str],
+        device_config_name: str | None = None,
+        max_concurrent_jobs: int = 100,
     ) -> cirq_google.ProcessorSampler:
         """Returns a sampler backed by the engine.
 
@@ -680,7 +685,10 @@ def get_sampler(
                 'to get_sampler() no longer supported. Use Engine.run() instead if '
                 'you need to specify a list.'
             )
-        return self.get_processor(processor_id).get_sampler(max_concurrent_jobs=max_concurrent_jobs)
+
+        return self.get_processor(processor_id).get_sampler(
+            device_config_name=device_config_name, max_concurrent_jobs=max_concurrent_jobs
+        )
 
     async def get_processor_config_from_snapshot_async(
         self, processor_id: str, snapshot_id: str, config_name: str = 'default'
diff --git a/cirq-google/cirq_google/engine/engine_processor_test.py b/cirq-google/cirq_google/engine/engine_processor_test.py
index 40dde396fc0..14e983dd436 100644
--- a/cirq-google/cirq_google/engine/engine_processor_test.py
+++ b/cirq-google/cirq_google/engine/engine_processor_test.py
@@ -346,23 +346,23 @@ def test_get_sampler_from_run_name() -> None:
     assert sampler.device_config_name == device_config_name
 
 
-def test_get_sampler_from_run_name_with_default_run_name() -> None:
+def test_get_sampler_from_run_name_with_default_values() -> None:
+    default_config_alias = 'test_alias'
     processor = cg.EngineProcessor(
         'a',
         'p',
         EngineContext(),
         _processor=quantum.QuantumProcessor(
             default_device_config_key=quantum.DeviceConfigKey(
-                run="run", config_alias="config_alias"
+                run="run", config_alias=default_config_alias
             )
         ),
     )
-    device_config_name = 'test_device_name'
 
-    sampler = processor.get_sampler_from_run_name(device_config_name=device_config_name)
+    sampler = processor.get_sampler_from_run_name()
 
     assert sampler.run_name == 'default'
-    assert sampler.device_config_name == device_config_name
+    assert sampler.device_config_name == default_config_alias
 
 
 def test_get_sampler_from_snapshot_id() -> None:
@@ -387,6 +387,28 @@ def test_get_sampler_from_snapshot_id() -> None:
     assert sampler.device_config_name == device_config_name
 
 
+def test_get_sampler_from_snapshot_id_with_default_device() -> None:
+    default_config_alias = 'test_alias'
+    processor = cg.EngineProcessor(
+        'a',
+        'p',
+        EngineContext(),
+        _processor=quantum.QuantumProcessor(
+            default_device_config_key=quantum.DeviceConfigKey(
+                run="run", config_alias=default_config_alias
+            )
+        ),
+    )
+    snapshot_id = 'test_snapshot'
+
+    sampler = processor.get_sampler_from_snapshot_id(
+        snapshot_id=snapshot_id, device_config_name=default_config_alias
+    )
+
+    assert sampler.snapshot_id == snapshot_id
+    assert sampler.device_config_name == default_config_alias
+
+
 def test_get_sampler_initializes_default_device_configuration() -> None:
     processor = cg.EngineProcessor(
         'a',
diff --git a/cirq-google/cirq_google/engine/engine_test.py b/cirq-google/cirq_google/engine/engine_test.py
index fcb77ff84e1..b7db8ec5011 100644
--- a/cirq-google/cirq_google/engine/engine_test.py
+++ b/cirq-google/cirq_google/engine/engine_test.py
@@ -815,6 +815,44 @@ def test_get_sampler_initializes_max_concurrent_jobs():
     assert sampler.max_concurrent_jobs == max_concurrent_jobs
 
 
+def test_get_sampler_from_run_name():
+    processor_id = 'test_processor_id'
+    run_name = 'test_run_name'
+    device_config_name = 'test_config_alias'
+    project_id = 'test_proj'
+    engine = cg.Engine(project_id=project_id)
+    processor = engine.get_processor(processor_id=processor_id)
+
+    processor_sampler = processor.get_sampler_from_run_name(
+        run_name=run_name, device_config_name=device_config_name
+    )
+    engine_sampler = engine.get_sampler_from_run_name(
+        processor_id=processor_id, run_name=run_name, device_config_name=device_config_name
+    )
+
+    assert processor_sampler.run_name == engine_sampler.run_name
+    assert processor_sampler.device_config_name == engine_sampler.device_config_name
+
+
+def test_get_sampler_from_snapshot():
+    processor_id = 'test_processor_id'
+    snapshot_id = 'test_snapshot_id'
+    device_config_name = 'test_config_alias'
+    project_id = 'test_proj'
+    engine = cg.Engine(project_id=project_id)
+    processor = engine.get_processor(processor_id=processor_id)
+
+    processor_sampler = processor.get_sampler_from_snapshot_id(
+        snapshot_id=snapshot_id, device_config_name=device_config_name
+    )
+    engine_sampler = engine.get_sampler_from_snapshot_id(
+        processor_id=processor_id, snapshot_id=snapshot_id, device_config_name=device_config_name
+    )
+
+    assert processor_sampler.snapshot_id == engine_sampler.snapshot_id
+    assert processor_sampler.device_config_name == engine_sampler.device_config_name
+
+
 @mock.patch('cirq_google.engine.engine_client.EngineClient', autospec=True)
 def test_sampler_with_unary_rpcs(client):
     setup_run_circuit_with_result_(client, _RESULTS)

From 95e1046fcb4fa5cab01c1b2fadf059e228f0ed05 Mon Sep 17 00:00:00 2001
From: dyates <devyates@google.com>
Date: Fri, 26 Sep 2025 18:12:02 +0000
Subject: [PATCH 06/12] Remove default run_name from functions

---
 cirq-google/cirq_google/engine/engine.py                | 2 +-
 cirq-google/cirq_google/engine/engine_processor.py      | 5 +----
 cirq-google/cirq_google/engine/engine_processor_test.py | 5 +++--
 3 files changed, 5 insertions(+), 7 deletions(-)

diff --git a/cirq-google/cirq_google/engine/engine.py b/cirq-google/cirq_google/engine/engine.py
index 29efb786d3b..de6be3f6874 100644
--- a/cirq-google/cirq_google/engine/engine.py
+++ b/cirq-google/cirq_google/engine/engine.py
@@ -593,8 +593,8 @@ def get_processor(self, processor_id: str) -> engine_processor.EngineProcessor:
     def get_sampler_from_run_name(
         self,
         processor_id: str,
+        run_name: str,
         device_config_name: str | None = None,
-        run_name: str = 'default',
         max_concurrent_jobs: int = 100,
     ) -> cirq_google.ProcessorSampler:
         """Returns a sampler backed by the engine and given `run_name`.
diff --git a/cirq-google/cirq_google/engine/engine_processor.py b/cirq-google/cirq_google/engine/engine_processor.py
index e955a235cdd..56798daaaaf 100644
--- a/cirq-google/cirq_google/engine/engine_processor.py
+++ b/cirq-google/cirq_google/engine/engine_processor.py
@@ -100,10 +100,7 @@ def engine(self) -> engine_base.Engine:
         return engine_base.Engine(self.project_id, context=self.context)
 
     def get_sampler_from_run_name(
-        self,
-        run_name: str = 'default',
-        device_config_name: str | None = None,
-        max_concurrent_jobs: int = 100,
+        self, run_name: str, device_config_name: str | None = None, max_concurrent_jobs: int = 100
     ) -> cg.engine.ProcessorSampler:
         """Returns a sampler backed by the engine and given `run_name`.
 
diff --git a/cirq-google/cirq_google/engine/engine_processor_test.py b/cirq-google/cirq_google/engine/engine_processor_test.py
index 14e983dd436..199ac7fbb49 100644
--- a/cirq-google/cirq_google/engine/engine_processor_test.py
+++ b/cirq-google/cirq_google/engine/engine_processor_test.py
@@ -358,10 +358,11 @@ def test_get_sampler_from_run_name_with_default_values() -> None:
             )
         ),
     )
+    run_name = 'test_run'
 
-    sampler = processor.get_sampler_from_run_name()
+    sampler = processor.get_sampler_from_run_name(run_name=run_name)
 
-    assert sampler.run_name == 'default'
+    assert sampler.run_name == run_name
     assert sampler.device_config_name == default_config_alias
 
 

From 158c77bb9e19327121cad6efab6748d5fcb058cd Mon Sep 17 00:00:00 2001
From: dyates <devyates@google.com>
Date: Fri, 26 Sep 2025 18:22:32 +0000
Subject: [PATCH 07/12] Make function param types consistent.

---
 cirq-google/cirq_google/engine/engine.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/cirq-google/cirq_google/engine/engine.py b/cirq-google/cirq_google/engine/engine.py
index de6be3f6874..4e8e5acf145 100644
--- a/cirq-google/cirq_google/engine/engine.py
+++ b/cirq-google/cirq_google/engine/engine.py
@@ -626,7 +626,7 @@ def get_sampler_from_snapshot_id(
         self,
         processor_id: str,
         snapshot_id: str,
-        device_config_name: str,
+        device_config_name: str | None,
         max_concurrent_jobs: int = 100,
     ) -> cirq_google.ProcessorSampler:
         """Returns a sampler backed by the engine.

From c0fe0adb289f8c6452019b3adfd553ade9a49790 Mon Sep 17 00:00:00 2001
From: dyates <devyates@google.com>
Date: Fri, 26 Sep 2025 18:27:44 +0000
Subject: [PATCH 08/12] Add param doc comment for `processor_id`

---
 cirq-google/cirq_google/engine/engine.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/cirq-google/cirq_google/engine/engine.py b/cirq-google/cirq_google/engine/engine.py
index 4e8e5acf145..c6fb85a20fb 100644
--- a/cirq-google/cirq_google/engine/engine.py
+++ b/cirq-google/cirq_google/engine/engine.py
@@ -600,6 +600,7 @@ def get_sampler_from_run_name(
         """Returns a sampler backed by the engine and given `run_name`.
 
         Args:
+            processor_id: String identifier of which processor should be used to sample.
             run_name: A unique identifier representing an automation run for the
                 processor. An Automation Run contains a collection of device
                 configurations for the processor.
@@ -631,6 +632,7 @@ def get_sampler_from_snapshot_id(
     ) -> cirq_google.ProcessorSampler:
         """Returns a sampler backed by the engine.
         Args:
+            processor_id: String identifier of which processor should be used to sample.
             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.

From 3c019ac857c5c0ce028790f297283465daef5bf4 Mon Sep 17 00:00:00 2001
From: dyates <devyates@google.com>
Date: Fri, 26 Sep 2025 18:33:52 +0000
Subject: [PATCH 09/12] Fix type error.

---
 .../cirq_google/engine/abstract_processor.py        | 13 ++-----------
 1 file changed, 2 insertions(+), 11 deletions(-)

diff --git a/cirq-google/cirq_google/engine/abstract_processor.py b/cirq-google/cirq_google/engine/abstract_processor.py
index 64cfe92f986..261f11c5993 100644
--- a/cirq-google/cirq_google/engine/abstract_processor.py
+++ b/cirq-google/cirq_google/engine/abstract_processor.py
@@ -180,17 +180,8 @@ async def run_sweep_async(
     run_sweep = duet.sync(run_sweep_async)
 
     @abc.abstractmethod
-    def get_sampler(self, run_name: str = "", device_config_name: str = "") -> cg.ProcessorSampler:
-        """Returns a sampler backed by the processor.
-
-        Args:
-            run_name: A unique identifier representing an automation run for the
-                processor. An Automation Run 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.
-        """
+    def get_sampler(self) -> cg.ProcessorSampler:
+        """Returns a sampler backed by the processor."""
 
     @abc.abstractmethod
     def engine(self) -> abstract_engine.AbstractEngine | None:

From 14762916630a0508b52e351c34278aea29526813 Mon Sep 17 00:00:00 2001
From: dyates <devyates@google.com>
Date: Fri, 17 Oct 2025 19:40:25 +0000
Subject: [PATCH 10/12] Remove get_sampler_from* functions and replace with new
 pattern.

---
 cirq-google/cirq_google/engine/engine.py      |  85 ++++-----------
 .../cirq_google/engine/engine_processor.py    | 102 +++++-------------
 .../engine/engine_processor_test.py           |  61 ++++-------
 cirq-google/cirq_google/engine/engine_test.py |  27 +++--
 4 files changed, 78 insertions(+), 197 deletions(-)

diff --git a/cirq-google/cirq_google/engine/engine.py b/cirq-google/cirq_google/engine/engine.py
index c6fb85a20fb..461b1750ae2 100644
--- a/cirq-google/cirq_google/engine/engine.py
+++ b/cirq-google/cirq_google/engine/engine.py
@@ -29,6 +29,7 @@
 import enum
 import random
 import string
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, TypeVar
 
 import duet
@@ -68,6 +69,19 @@ class ProtoVersion(enum.Enum):
     V2 = 2
 
 
+@dataclass
+class Snapshot:
+    id: str
+
+
+@dataclass
+class Run:
+    id: str
+
+
+type DeviceVersion = Snapshot | Run
+
+
 def _make_random_id(prefix: str, length: int = 16):
     random_digits = [random.choice(string.ascii_uppercase + string.digits) for _ in range(length)]
     suffix = ''.join(random_digits)
@@ -590,75 +604,11 @@ def get_processor(self, processor_id: str) -> engine_processor.EngineProcessor:
         """
         return engine_processor.EngineProcessor(self.project_id, processor_id, self.context)
 
-    def get_sampler_from_run_name(
-        self,
-        processor_id: str,
-        run_name: str,
-        device_config_name: str | None = None,
-        max_concurrent_jobs: int = 100,
-    ) -> cirq_google.ProcessorSampler:
-        """Returns a sampler backed by the engine and given `run_name`.
-
-        Args:
-            processor_id: String identifier of which processor should be used to sample.
-            run_name: A unique identifier representing an automation run for the
-                processor. An Automation Run 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.
-            max_concurrent_jobs: The maximum number of jobs to be sent
-                simultaneously to the Engine. This client-side throttle can be
-                used to proactively reduce load to the backends and avoid quota
-                violations when pipelining circuit executions.
-
-        Returns:
-            A `cirq.Sampler` instance (specifically a `engine_sampler.ProcessorSampler`
-            that will send circuits to the Quantum Computing Service
-            when sampled.
-        """
-        return self.get_processor(processor_id).get_sampler_from_run_name(
-            run_name=run_name,
-            device_config_name=device_config_name,
-            max_concurrent_jobs=max_concurrent_jobs,
-        )
-
-    def get_sampler_from_snapshot_id(
-        self,
-        processor_id: str,
-        snapshot_id: str,
-        device_config_name: str | None,
-        max_concurrent_jobs: int = 100,
-    ) -> cirq_google.ProcessorSampler:
-        """Returns a sampler backed by the engine.
-        Args:
-            processor_id: String identifier of which processor should be used to sample.
-            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.
-            snapshot_id: A unique identifier for an immutable snapshot reference.
-                A snapshot contains a collection of device configurations for the
-                processor.
-            max_concurrent_jobs: The maximum number of jobs to be sent
-                simultaneously to the Engine. This client-side throttle can be
-                used to proactively reduce load to the backends and avoid quota
-                violations when pipelining circuit executions.
-
-        Returns:
-            A `cirq.Sampler` instance (specifically a `engine_sampler.ProcessorSampler`
-            that will send circuits to the Quantum Computing Service
-            when sampled.
-        """
-        return self.get_processor(processor_id).get_sampler_from_snapshot_id(
-            snapshot_id=snapshot_id,
-            device_config_name=device_config_name,
-            max_concurrent_jobs=max_concurrent_jobs,
-        )
-
     def get_sampler(
         self,
         processor_id: str | list[str],
         device_config_name: str | None = None,
+        device_version: DeviceVersion | None = None,
         max_concurrent_jobs: int = 100,
     ) -> cirq_google.ProcessorSampler:
         """Returns a sampler backed by the engine.
@@ -668,6 +618,7 @@ def get_sampler(
             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.
+            device_version: Specifies either the snapshot_id or the run_name.
             max_concurrent_jobs: The maximum number of jobs to be sent
                 concurrently to the Engine. This client-side throttle can be
                 used to proactively reduce load to the backends and avoid quota
@@ -689,7 +640,9 @@ def get_sampler(
             )
 
         return self.get_processor(processor_id).get_sampler(
-            device_config_name=device_config_name, max_concurrent_jobs=max_concurrent_jobs
+            device_config_name=device_config_name,
+            device_version=device_version,
+            max_concurrent_jobs=max_concurrent_jobs,
         )
 
     async def get_processor_config_from_snapshot_async(
diff --git a/cirq-google/cirq_google/engine/engine_processor.py b/cirq-google/cirq_google/engine/engine_processor.py
index 56798daaaaf..26ccf6013f6 100644
--- a/cirq-google/cirq_google/engine/engine_processor.py
+++ b/cirq-google/cirq_google/engine/engine_processor.py
@@ -17,6 +17,7 @@
 import datetime
 from typing import Any, TYPE_CHECKING
 
+import cirq_google.engine.engine as engine_base
 from cirq import _compat
 from cirq_google.api import v2
 from cirq_google.devices import grid_device
@@ -35,7 +36,6 @@
     import cirq_google as cg
     import cirq_google.cloud.quantum as quantum
     import cirq_google.engine.abstract_job as abstract_job
-    import cirq_google.engine.engine as engine_base
 
 
 def _date_to_timestamp(union_time: datetime.datetime | datetime.date | int | None) -> int | None:
@@ -99,78 +99,11 @@ def engine(self) -> engine_base.Engine:
 
         return engine_base.Engine(self.project_id, context=self.context)
 
-    def get_sampler_from_run_name(
-        self, run_name: str, device_config_name: str | None = None, max_concurrent_jobs: int = 100
-    ) -> cg.engine.ProcessorSampler:
-        """Returns a sampler backed by the engine and given `run_name`.
-
-        Args:
-            run_name: A unique identifier representing an automation run for the
-                processor. An Automation Run 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.
-            max_concurrent_jobs: The maximum number of jobs to be sent
-                simultaneously to the Engine. This client-side throttle can be
-                used to proactively reduce load to the backends and avoid quota
-                violations when pipelining circuit executions.
-
-        Returns:
-            A `cirq.Sampler` instance (specifically a `engine_sampler.ProcessorSampler`
-            that will send circuits to the Quantum Computing Service
-            when sampled.
-        """
-        processor = self._inner_processor()
-        return processor_sampler.ProcessorSampler(
-            processor=self,
-            run_name=run_name,
-            device_config_name=(
-                device_config_name
-                if device_config_name
-                else processor.default_device_config_key.config_alias
-            ),
-            max_concurrent_jobs=max_concurrent_jobs,
-        )
-
-    def get_sampler_from_snapshot_id(
+    def get_sampler(
         self,
-        snapshot_id: str,
         device_config_name: str | None = None,
+        device_version: engine_base.DeviceVersion | None = None,
         max_concurrent_jobs: int = 100,
-    ) -> cg.engine.ProcessorSampler:
-        """Returns a sampler backed by the engine.
-        Args:
-            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.
-            snapshot_id: A unique identifier for an immutable snapshot reference.
-                A snapshot contains a collection of device configurations for the
-                processor.
-            max_concurrent_jobs: The maximum number of jobs to be sent
-                simultaneously to the Engine. This client-side throttle can be
-                used to proactively reduce load to the backends and avoid quota
-                violations when pipelining circuit executions.
-
-        Returns:
-            A `cirq.Sampler` instance (specifically a `engine_sampler.ProcessorSampler`
-            that will send circuits to the Quantum Computing Service
-            when sampled.
-        """
-        processor = self._inner_processor()
-        return processor_sampler.ProcessorSampler(
-            processor=self,
-            snapshot_id=snapshot_id,
-            device_config_name=(
-                device_config_name
-                if device_config_name
-                else processor.default_device_config_key.config_alias
-            ),
-            max_concurrent_jobs=max_concurrent_jobs,
-        )
-
-    def get_sampler(
-        self, device_config_name: str | None = None, max_concurrent_jobs: int = 100
     ) -> cg.engine.ProcessorSampler:
         """Returns the default sampler backed by the engine.
 
@@ -178,6 +111,7 @@ def get_sampler(
             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.
+            device_version: Specifies either the snapshot_id or the run_name.
             max_concurrent_jobs: The maximum number of jobs to be sent
                 simultaneously to the Engine. This client-side throttle can be
                 used to proactively reduce load to the backends and avoid quota
@@ -191,15 +125,31 @@ def get_sampler(
         """
         processor = self._inner_processor()
 
+        device_config_name = (
+            device_config_name
+            if device_config_name
+            else processor.default_device_config_key.config_alias
+        )
+
+        if isinstance(device_version, engine_base.Snapshot):
+            return processor_sampler.ProcessorSampler(
+                processor=self,
+                snapshot_id=device_version.id,
+                device_config_name=device_config_name,
+                max_concurrent_jobs=max_concurrent_jobs,
+            )
+        if isinstance(device_version, engine_base.Run):
+            return processor_sampler.ProcessorSampler(
+                processor=self,
+                run_name=device_version.id,
+                device_config_name=device_config_name,
+                max_concurrent_jobs=max_concurrent_jobs,
+            )
+
         return processor_sampler.ProcessorSampler(
             processor=self,
             run_name=processor.default_device_config_key.run,
-            snapshot_id=processor.default_device_config_key.snapshot_id,
-            device_config_name=(
-                device_config_name
-                if device_config_name
-                else processor.default_device_config_key.config_alias
-            ),
+            device_config_name=device_config_name,
             max_concurrent_jobs=max_concurrent_jobs,
         )
 
diff --git a/cirq-google/cirq_google/engine/engine_processor_test.py b/cirq-google/cirq_google/engine/engine_processor_test.py
index 1a98d9c4244..035a6d8f1e9 100644
--- a/cirq-google/cirq_google/engine/engine_processor_test.py
+++ b/cirq-google/cirq_google/engine/engine_processor_test.py
@@ -30,7 +30,7 @@
 from cirq_google.api import v2
 from cirq_google.cloud import quantum
 from cirq_google.engine import engine_client, ProcessorConfig, util
-from cirq_google.engine.engine import EngineContext
+from cirq_google.engine.engine import EngineContext, Run, Snapshot
 
 
 def _to_timestamp(json_string):
@@ -335,19 +335,17 @@ def test_get_sampler_from_run_name() -> None:
             )
         ),
     )
-    run_name = 'test_run_name'
+    run = Run(id='test_run_name')
     device_config_name = 'test_device_name'
 
-    sampler = processor.get_sampler_from_run_name(
-        run_name=run_name, device_config_name=device_config_name
-    )
+    sampler = processor.get_sampler(device_version=run, device_config_name=device_config_name)
 
-    assert sampler.run_name == run_name
+    assert sampler.run_name == run.id
     assert sampler.device_config_name == device_config_name
 
 
-def test_get_sampler_from_run_name_with_default_values() -> None:
-    default_config_alias = 'test_alias'
+def test_get_sampler_from_run_name_with_defaults() -> None:
+    default_config_alias = 'default_alias'
     processor = cg.EngineProcessor(
         'a',
         'p',
@@ -358,75 +356,56 @@ def test_get_sampler_from_run_name_with_default_values() -> None:
             )
         ),
     )
-    run_name = 'test_run'
+    run = Run(id='test_run_name')
 
-    sampler = processor.get_sampler_from_run_name(run_name=run_name)
+    sampler = processor.get_sampler(device_version=run)
 
-    assert sampler.run_name == run_name
+    assert sampler.run_name == run.id
     assert sampler.device_config_name == default_config_alias
 
 
 def test_get_sampler_from_snapshot_id() -> None:
+    default_snapshot_id = 'default_snap'
     processor = cg.EngineProcessor(
         'a',
         'p',
         EngineContext(),
         _processor=quantum.QuantumProcessor(
             default_device_config_key=quantum.DeviceConfigKey(
-                run="run", config_alias="config_alias"
+                config_alias="config_alias", snapshot_id=default_snapshot_id
             )
         ),
     )
-    snapshot_id = 'test_snapshot'
+    snapshot = Snapshot(id='test_snapshot')
     device_config_name = 'test_device_name'
 
-    sampler = processor.get_sampler_from_snapshot_id(
-        snapshot_id=snapshot_id, device_config_name=device_config_name
-    )
+    sampler = processor.get_sampler(device_version=snapshot, device_config_name=device_config_name)
 
-    assert sampler.snapshot_id == snapshot_id
+    assert sampler.snapshot_id == snapshot.id
     assert sampler.device_config_name == device_config_name
 
 
-def test_get_sampler_from_snapshot_id_with_default_device() -> None:
+def test_get_sampler_from_snapshot_id_with_defaults() -> None:
     default_config_alias = 'test_alias'
+    default_snapshot_id = 'default_snapshot'
     processor = cg.EngineProcessor(
         'a',
         'p',
         EngineContext(),
         _processor=quantum.QuantumProcessor(
             default_device_config_key=quantum.DeviceConfigKey(
-                run="run", config_alias=default_config_alias
+                config_alias=default_config_alias, snapshot_id=default_snapshot_id
             )
         ),
     )
-    snapshot_id = 'test_snapshot'
+    snapshot = Snapshot(id='test_snapshot')
 
-    sampler = processor.get_sampler_from_snapshot_id(
-        snapshot_id=snapshot_id, device_config_name=default_config_alias
-    )
+    sampler = processor.get_sampler(device_version=snapshot)
 
-    assert sampler.snapshot_id == snapshot_id
+    assert sampler.snapshot_id == snapshot.id
     assert sampler.device_config_name == default_config_alias
 
 
-def test_get_sampler_initializes_default_device_configuration() -> None:
-    processor = cg.EngineProcessor(
-        'a',
-        'p',
-        EngineContext(),
-        _processor=quantum.QuantumProcessor(
-            default_device_config_key=quantum.DeviceConfigKey(
-                run="run", config_alias="config_alias"
-            )
-        ),
-    )
-    sampler = processor.get_sampler()
-
-    assert sampler.run_name == "run"
-    assert sampler.device_config_name == "config_alias"
-
-
 @mock.patch('cirq_google.engine.engine_client.EngineClient.get_processor_async')
 def test_get_sampler_loads_processor_with_default_device_configuration(get_processor) -> None:
     get_processor.return_value = quantum.QuantumProcessor(
diff --git a/cirq-google/cirq_google/engine/engine_test.py b/cirq-google/cirq_google/engine/engine_test.py
index b7db8ec5011..2d1474bf329 100644
--- a/cirq-google/cirq_google/engine/engine_test.py
+++ b/cirq-google/cirq_google/engine/engine_test.py
@@ -32,7 +32,7 @@
 from cirq_google.api import v1, v2
 from cirq_google.cloud import quantum
 from cirq_google.engine import util
-from cirq_google.engine.engine import EngineContext
+from cirq_google.engine.engine import EngineContext, Run, Snapshot
 
 _CIRCUIT = cirq.Circuit(
     cirq.X(cirq.GridQubit(5, 2)) ** 0.5, cirq.measure(cirq.GridQubit(5, 2), key='result')
@@ -296,9 +296,8 @@ def test_engine_get_sampler_with_snapshot_id_passes_to_unary_rpc(client):
         project_id='proj',
         context=EngineContext(service_args={'client_info': 1}, enable_streaming=False),
     )
-    sampler = engine.get_sampler_from_snapshot_id(
-        'mysim', device_config_name="config", snapshot_id="123"
-    )
+    snapshot_id = Snapshot(id="123")
+    sampler = engine.get_sampler('mysim', device_config_name="config", device_version=snapshot_id)
     _ = sampler.run_sweep(_CIRCUIT, params=[cirq.ParamResolver({'a': 1})])
 
     kwargs = client().create_job_async.call_args_list[0].kwargs
@@ -817,17 +816,17 @@ def test_get_sampler_initializes_max_concurrent_jobs():
 
 def test_get_sampler_from_run_name():
     processor_id = 'test_processor_id'
-    run_name = 'test_run_name'
+    run = Run(id="test_run_name")
     device_config_name = 'test_config_alias'
     project_id = 'test_proj'
     engine = cg.Engine(project_id=project_id)
     processor = engine.get_processor(processor_id=processor_id)
 
-    processor_sampler = processor.get_sampler_from_run_name(
-        run_name=run_name, device_config_name=device_config_name
+    processor_sampler = processor.get_sampler(
+        device_version=run, device_config_name=device_config_name
     )
-    engine_sampler = engine.get_sampler_from_run_name(
-        processor_id=processor_id, run_name=run_name, device_config_name=device_config_name
+    engine_sampler = engine.get_sampler(
+        processor_id=processor_id, device_version=run, device_config_name=device_config_name
     )
 
     assert processor_sampler.run_name == engine_sampler.run_name
@@ -836,17 +835,17 @@ def test_get_sampler_from_run_name():
 
 def test_get_sampler_from_snapshot():
     processor_id = 'test_processor_id'
-    snapshot_id = 'test_snapshot_id'
+    snapshot_id = Snapshot(id='test_snapshot_id')
     device_config_name = 'test_config_alias'
     project_id = 'test_proj'
     engine = cg.Engine(project_id=project_id)
     processor = engine.get_processor(processor_id=processor_id)
 
-    processor_sampler = processor.get_sampler_from_snapshot_id(
-        snapshot_id=snapshot_id, device_config_name=device_config_name
+    processor_sampler = processor.get_sampler(
+        device_version=snapshot_id, device_config_name=device_config_name
     )
-    engine_sampler = engine.get_sampler_from_snapshot_id(
-        processor_id=processor_id, snapshot_id=snapshot_id, device_config_name=device_config_name
+    engine_sampler = engine.get_sampler(
+        processor_id=processor_id, device_config_name=device_config_name, device_version=snapshot_id
     )
 
     assert processor_sampler.snapshot_id == engine_sampler.snapshot_id

From 4849b9ae169276062721d23236b9e9a9be303824 Mon Sep 17 00:00:00 2001
From: dyates <devyates@google.com>
Date: Fri, 24 Oct 2025 20:31:36 +0000
Subject: [PATCH 11/12] Move `DeviceVersion` to processor_config. Remove all
 `*from_config/snapshot` functions.

---
 cirq-google/cirq_google/engine/engine.py      |  63 ++-----
 .../cirq_google/engine/engine_client.py       | 106 ++++-------
 .../cirq_google/engine/engine_client_test.py  | 108 +++++-------
 .../cirq_google/engine/engine_processor.py    |  41 ++---
 .../engine/engine_processor_test.py           | 166 +++++++-----------
 cirq-google/cirq_google/engine/engine_test.py |  63 +++----
 .../cirq_google/engine/processor_config.py    |  52 +++++-
 .../engine/processor_config_test.py           |  53 ++++--
 8 files changed, 278 insertions(+), 374 deletions(-)

diff --git a/cirq-google/cirq_google/engine/engine.py b/cirq-google/cirq_google/engine/engine.py
index 461b1750ae2..0e33ebb7fe2 100644
--- a/cirq-google/cirq_google/engine/engine.py
+++ b/cirq-google/cirq_google/engine/engine.py
@@ -29,8 +29,7 @@
 import enum
 import random
 import string
-from dataclasses import dataclass
-from typing import TYPE_CHECKING, TypeVar
+from typing import TYPE_CHECKING, TypeAlias, TypeVar
 
 import duet
 import google.auth
@@ -69,19 +68,6 @@ class ProtoVersion(enum.Enum):
     V2 = 2
 
 
-@dataclass
-class Snapshot:
-    id: str
-
-
-@dataclass
-class Run:
-    id: str
-
-
-type DeviceVersion = Snapshot | Run
-
-
 def _make_random_id(prefix: str, length: int = 16):
     random_digits = [random.choice(string.ascii_uppercase + string.digits) for _ in range(length)]
     suffix = ''.join(random_digits)
@@ -608,7 +594,7 @@ def get_sampler(
         self,
         processor_id: str | list[str],
         device_config_name: str | None = None,
-        device_version: DeviceVersion | None = None,
+        device_version: processor_config.DeviceVersion | None = None,
         max_concurrent_jobs: int = 100,
     ) -> cirq_google.ProcessorSampler:
         """Returns a sampler backed by the engine.
@@ -645,34 +631,11 @@ def get_sampler(
             max_concurrent_jobs=max_concurrent_jobs,
         )
 
-    async def get_processor_config_from_snapshot_async(
-        self, processor_id: str, snapshot_id: str, config_name: str = 'default'
-    ) -> processor_config.ProcessorConfig | None:
-        """Returns a ProcessorConfig from this project and the given processor id.
-
-        Args:
-           processor_id: The processor unique identifier.
-           snapshot_id: The unique identifier for the snapshot.
-           config_name: The identifier for the config.
-
-        Returns:
-           The ProcessorConfig from this project and processor.
-        """
-        client = self.context.client
-        quantum_config = await client.get_quantum_processor_config_from_snapshot_async(
-            project_id=self.project_id,
-            processor_id=processor_id,
-            snapshot_id=snapshot_id,
-            config_name=config_name,
-        )
-        if quantum_config:
-            return processor_config.ProcessorConfig(quantum_processor_config=quantum_config)
-        return None
-
-    get_processor_config_from_snapshot = duet.sync(get_processor_config_from_snapshot_async)
-
-    async def get_processor_config_from_run_async(
-        self, processor_id: str, run_name: str = 'current', config_name: str = 'default'
+    async def get_processor_config_async(
+        self,
+        processor_id: str,
+        device_version: processor_config.DeviceVersion = processor_config.Run(id='current'),
+        config_name: str = 'default',
     ) -> processor_config.ProcessorConfig | None:
         """Returns a ProcessorConfig from this project and the given processor id.
 
@@ -681,25 +644,27 @@ async def get_processor_config_from_run_async(
 
         Args:
             processor_id: The processor unique identifier.
-            run_name: The unique identifier for the automation run.
+            device_version: Specifies either the snapshot_id or the run_name.
             config_name: The identifier for the config.
 
         Returns:
             The ProcessorConfig from this project and processor.
         """
-        quantum_config = await self.context.client.get_quantum_processor_config_from_run_async(
+        quantum_config = await self.context.client.get_quantum_processor_config_async(
             project_id=self.project_id,
             processor_id=processor_id,
-            run_name=run_name,
+            device_version=device_version,
             config_name=config_name,
         )
         if quantum_config:
             return processor_config.ProcessorConfig(
-                quantum_processor_config=quantum_config, run_name=run_name
+                processor=self.get_processor(processor_id),
+                quantum_processor_config=quantum_config,
+                device_version=device_version,
             )
         return None
 
-    get_processor_config_from_run = duet.sync(get_processor_config_from_run_async)
+    get_processor_config = duet.sync(get_processor_config_async)
 
 
 def get_engine(project_id: str | None = None) -> Engine:
diff --git a/cirq-google/cirq_google/engine/engine_client.py b/cirq-google/cirq_google/engine/engine_client.py
index baa32be5239..de5df2fc3ca 100644
--- a/cirq-google/cirq_google/engine/engine_client.py
+++ b/cirq-google/cirq_google/engine/engine_client.py
@@ -30,6 +30,7 @@
 from cirq_google.cloud import quantum
 from cirq_google.engine import stream_manager
 from cirq_google.engine.asyncio_executor import AsyncioExecutor
+from cirq_google.engine.processor_config import DeviceVersion, Run, Snapshot
 
 _M = TypeVar('_M', bound=proto.Message)
 _R = TypeVar('_R')
@@ -1181,29 +1182,19 @@ async def list_time_slots_async(
 
     list_time_slots = duet.sync(list_time_slots_async)
 
-    async def _get_quantum_processor_config(
-        self, name: str
-    ) -> quantum.QuantumProcessorConfig | None:
-        """Runs get_quantum_processor_config with the given resource name."""
-        try:
-            request = quantum.GetQuantumProcessorConfigRequest(name=name)
-            return await self._send_request_async(
-                self.grpc_client.get_quantum_processor_config, request
-            )
-        except EngineException as err:
-            if isinstance(err.__cause__, NotFound):
-                return None
-            raise
-
-    async def get_quantum_processor_config_from_snapshot_async(
-        self, project_id: str, processor_id: str, snapshot_id: str, config_name: str
+    async def get_quantum_processor_config_async(
+        self,
+        project_id: str,
+        processor_id: str,
+        config_name: str = 'default',
+        device_version: DeviceVersion = Run(id='current'),
     ) -> quantum.QuantumProcessorConfig | None:
         """Returns the QuantumProcessorConfig for the given snapshot id.
 
         Args:
             project_id: A project_id of the parent Google Cloud Project.
             processor_id: The processor unique identifier.
-            snapshot_id: The id of the snapshot that contains the quantum processor config.
+            device_version: Specifies either the snapshot_id or the run_name.
             config_name: The id of the quantum processor config.
 
         Returns:
@@ -1212,44 +1203,23 @@ async def get_quantum_processor_config_from_snapshot_async(
         Raises:
             EngineException: If the request to get the config fails.
         """
-        name = _quantum_processor_config_name_from_snapshot_id(
-            project_id=project_id,
-            processor_id=processor_id,
-            snapshot_id=snapshot_id,
-            config_name=config_name,
-        )
-        return await self._get_quantum_processor_config(name)
-
-    get_quantum_processor_config_from_snapshot = duet.sync(
-        get_quantum_processor_config_from_snapshot_async
-    )
-
-    async def get_quantum_processor_config_from_run_async(
-        self, project_id: str, processor_id: str, run_name: str, config_name: str
-    ) -> quantum.QuantumProcessorConfig | None:
-        """Returns the QuantumProcessorConfig for the given run_name.
-
-        Args:
-            project_id: A project_id of the parent Google Cloud Project.
-            processor_id: The processor unique identifier.
-            config_name: The id of the quantum processor config.
-            run_name: The run_name that contains the quantum processor config.
-
-        Returns:
-            The quantum procesor config or None if it does not exist.
-
-        Raises:
-            EngineException: If the request to get the config fails.
-        """
-        name = _quantum_processor_config_name_from_run_name(
-            project_id=project_id,
-            processor_id=processor_id,
-            run_name=run_name,
-            config_name=config_name,
-        )
-        return await self._get_quantum_processor_config(name)
+        try:
+            name = _quantum_processor_config_name_from_device_version(
+                project_id=project_id,
+                processor_id=processor_id,
+                config_name=config_name,
+                device_version=device_version,
+            )
+            request = quantum.GetQuantumProcessorConfigRequest(name=name)
+            return await self._send_request_async(
+                self.grpc_client.get_quantum_processor_config, request
+            )
+        except EngineException as err:
+            if isinstance(err.__cause__, NotFound):
+                return None
+            raise
 
-    get_quantum_processor_config_from_run = duet.sync(get_quantum_processor_config_from_run_async)
+    get_quantum_processor_config = duet.sync(get_quantum_processor_config_async)
 
 
 def _project_name(project_id: str) -> str:
@@ -1300,22 +1270,22 @@ def _ids_from_calibration_name(calibration_name: str) -> tuple[str, str, int]:
     return parts[1], parts[3], int(parts[5])
 
 
-def _quantum_processor_config_name_from_snapshot_id(
-    project_id: str, processor_id: str, snapshot_id: str, config_name: str
-) -> str:
-    return (
-        f'{_processor_name_from_ids(project_id, processor_id)}/'
-        f'configSnapshots/{snapshot_id}/'
-        f'configs/{config_name}'
-    )
-
-
-def _quantum_processor_config_name_from_run_name(
-    project_id: str, processor_id: str, run_name: str, config_name: str
+def _quantum_processor_config_name_from_device_version(
+    project_id: str,
+    processor_id: str,
+    config_name: str,
+    device_version: DeviceVersion | None = None,
 ) -> str:
+    processor_resource_name = _processor_name_from_ids(project_id, processor_id)
+    if isinstance(device_version, Snapshot):
+        return (
+            f'{processor_resource_name}/'
+            f'configSnapshots/{device_version.id}/'
+            f'configs/{config_name}'
+        )
     return (
-        f'{_processor_name_from_ids(project_id, processor_id)}/'
-        f'configAutomationRuns/{run_name}/'
+        f'{processor_resource_name}/'
+        f'configAutomationRuns/{device_version.id if device_version else 'default'}/'
         f'configs/{config_name}'
     )
 
diff --git a/cirq-google/cirq_google/engine/engine_client_test.py b/cirq-google/cirq_google/engine/engine_client_test.py
index 61fa5b5f3b0..85dfee86435 100644
--- a/cirq-google/cirq_google/engine/engine_client_test.py
+++ b/cirq-google/cirq_google/engine/engine_client_test.py
@@ -32,6 +32,7 @@
 import cirq_google.engine.stream_manager as engine_stream_manager
 from cirq_google.cloud import quantum
 from cirq_google.engine.engine_client import EngineClient, EngineException
+from cirq_google.engine.processor_config import Run, Snapshot
 
 # JOB_PATH represents the path to a specific job.
 JOB_PATH = 'projects/proj/programs/prog/jobs/job0'
@@ -1747,15 +1748,14 @@ def test_list_time_slots(client_constructor, default_engine_client):
 
 @mock.patch.object(quantum, 'QuantumEngineServiceAsyncClient', autospec=True)
 def test_get_quantum_processor_config_from_snapshot(client_constructor, default_engine_client):
-
     project_id = "test_project_id"
     processor_id = "test_processor_id"
-    snapshot_id = "test_snapshot_id"
+    snapshot = Snapshot(id="test_snapshot_id")
     config_name = "test_config_name"
     resource_name = (
         f'projects/{project_id}/'
         f'processors/{processor_id}/'
-        f'configSnapshots/{snapshot_id}/'
+        f'configSnapshots/{snapshot.id}/'
         f'configs/{config_name}'
     )
 
@@ -1763,70 +1763,73 @@ def test_get_quantum_processor_config_from_snapshot(client_constructor, default_
     expected_result = quantum.QuantumProcessorConfig(name=resource_name)
     grpc_client.get_quantum_processor_config.return_value = expected_result
 
-    actual_result = default_engine_client.get_quantum_processor_config_from_snapshot(
-        project_id=project_id,
-        processor_id=processor_id,
-        config_name=config_name,
-        snapshot_id=snapshot_id,
+    assert (
+        default_engine_client.get_quantum_processor_config(
+            project_id=project_id,
+            processor_id=processor_id,
+            config_name=config_name,
+            device_version=snapshot,
+        )
+        == expected_result
     )
     grpc_client.get_quantum_processor_config.assert_called_with(
         quantum.GetQuantumProcessorConfigRequest(name=resource_name)
     )
-    assert actual_result == expected_result
 
 
 @mock.patch.object(quantum, 'QuantumEngineServiceAsyncClient', autospec=True)
-def test_get_quantum_processor_config_from_snapshot_not_found(
-    client_constructor, default_engine_client
-):
+def test_get_quantum_processor_exception(client_constructor, default_engine_client):
+    grpc_client = _setup_client_mock(client_constructor)
+    grpc_client.get_quantum_processor_config.side_effect = exceptions.BadRequest('invalid_reueust')
+
+    with pytest.raises(EngineException, match='invalid_reueust'):
+        _ = default_engine_client.get_quantum_processor_config(
+            project_id="test_project_id",
+            processor_id="test_processor_id",
+            config_name="test_config_name",
+        )
+
+
+@mock.patch.object(quantum, 'QuantumEngineServiceAsyncClient', autospec=True)
+def test_get_quantum_processor_config_from_run(client_constructor, default_engine_client):
+
     project_id = "test_project_id"
     processor_id = "test_processor_id"
-    snapshot_id = "test_snapshot_id"
+    run = Run(id="test_run_name")
     config_name = "test_config_name"
     resource_name = (
         f'projects/{project_id}/'
         f'processors/{processor_id}/'
-        f'configSnapshots/{snapshot_id}/'
+        f'configAutomationRuns/{run.id}/'
         f'configs/{config_name}'
     )
 
     grpc_client = _setup_client_mock(client_constructor)
-    grpc_client.get_quantum_processor_config.side_effect = exceptions.NotFound('not found')
+    expected_result = quantum.QuantumProcessorConfig(name=resource_name)
+    grpc_client.get_quantum_processor_config.return_value = expected_result
 
-    actual_result = default_engine_client.get_quantum_processor_config_from_snapshot(
-        project_id=project_id,
-        processor_id=processor_id,
-        config_name=config_name,
-        snapshot_id=snapshot_id,
+    assert (
+        default_engine_client.get_quantum_processor_config(
+            project_id=project_id,
+            processor_id=processor_id,
+            config_name=config_name,
+            device_version=run,
+        )
+        == expected_result
     )
     grpc_client.get_quantum_processor_config.assert_called_with(
         quantum.GetQuantumProcessorConfigRequest(name=resource_name)
     )
-    assert actual_result is None
 
 
 @mock.patch.object(quantum, 'QuantumEngineServiceAsyncClient', autospec=True)
-def test_get_quantum_processor_config_from_snapshot_exception(
+def test_get_quantum_processor_config_defaults_to_current_run(
     client_constructor, default_engine_client
 ):
-    grpc_client = _setup_client_mock(client_constructor)
-    grpc_client.get_quantum_processor_config.side_effect = exceptions.BadRequest('invalid_reueust')
-
-    with pytest.raises(EngineException, match='invalid_reueust'):
-        _ = default_engine_client.get_quantum_processor_config_from_snapshot(
-            project_id="test_project_id",
-            processor_id="test_processor_id",
-            config_name="test_config_name",
-            snapshot_id="test_snapshot_id",
-        )
-
-
-@mock.patch.object(quantum, 'QuantumEngineServiceAsyncClient', autospec=True)
-def test_get_quantum_processor_config_from_run(client_constructor, default_engine_client):
 
     project_id = "test_project_id"
     processor_id = "test_processor_id"
-    run_name = "test_run_name"
+    run_name = 'current'
     config_name = "test_config_name"
     resource_name = (
         f'projects/{project_id}/'
@@ -1836,51 +1839,34 @@ def test_get_quantum_processor_config_from_run(client_constructor, default_engin
     )
 
     grpc_client = _setup_client_mock(client_constructor)
-    expected_result = quantum.QuantumProcessorConfig(name=resource_name)
-    grpc_client.get_quantum_processor_config.return_value = expected_result
+    grpc_client.get_quantum_processor_config.return_value = None
 
-    actual_result = default_engine_client.get_quantum_processor_config_from_run(
-        project_id=project_id, processor_id=processor_id, config_name=config_name, run_name=run_name
+    default_engine_client.get_quantum_processor_config(
+        project_id=project_id, processor_id=processor_id, config_name=config_name
     )
     grpc_client.get_quantum_processor_config.assert_called_with(
         quantum.GetQuantumProcessorConfigRequest(name=resource_name)
     )
-    assert actual_result == expected_result
 
 
 @mock.patch.object(quantum, 'QuantumEngineServiceAsyncClient', autospec=True)
-def test_get_quantum_processor_config_from_run_not_found(client_constructor, default_engine_client):
+def test_get_quantum_processor_config_not_found(client_constructor, default_engine_client):
     project_id = "test_project_id"
     processor_id = "test_processor_id"
-    run_name = "test_run_name"
     config_name = "test_config_name"
     resource_name = (
         f'projects/{project_id}/'
         f'processors/{processor_id}/'
-        f'configAutomationRuns/{run_name}/'
+        'configAutomationRuns/current/'
         f'configs/{config_name}'
     )
     grpc_client = _setup_client_mock(client_constructor)
     grpc_client.get_quantum_processor_config.side_effect = exceptions.NotFound('not found')
 
-    actual_result = default_engine_client.get_quantum_processor_config_from_run(
-        project_id=project_id, processor_id=processor_id, config_name=config_name, run_name=run_name
+    actual_result = default_engine_client.get_quantum_processor_config(
+        project_id=project_id, processor_id=processor_id, config_name=config_name
     )
     grpc_client.get_quantum_processor_config.assert_called_with(
         quantum.GetQuantumProcessorConfigRequest(name=resource_name)
     )
     assert actual_result is None
-
-
-@mock.patch.object(quantum, 'QuantumEngineServiceAsyncClient', autospec=True)
-def test_get_quantum_processor_config_from_run_exception(client_constructor, default_engine_client):
-    grpc_client = _setup_client_mock(client_constructor)
-    grpc_client.get_quantum_processor_config.side_effect = exceptions.BadRequest('invalid_reueust')
-
-    with pytest.raises(EngineException, match='invalid_reueust'):
-        _ = default_engine_client.get_quantum_processor_config_from_run(
-            project_id="test_project_id",
-            processor_id="test_processor_id",
-            config_name="test_config_name",
-            run_name="test_run_name",
-        )
diff --git a/cirq-google/cirq_google/engine/engine_processor.py b/cirq-google/cirq_google/engine/engine_processor.py
index 26ccf6013f6..80395d4cb38 100644
--- a/cirq-google/cirq_google/engine/engine_processor.py
+++ b/cirq-google/cirq_google/engine/engine_processor.py
@@ -102,7 +102,7 @@ def engine(self) -> engine_base.Engine:
     def get_sampler(
         self,
         device_config_name: str | None = None,
-        device_version: engine_base.DeviceVersion | None = None,
+        device_version: processor_config.DeviceVersion | None = None,
         max_concurrent_jobs: int = 100,
     ) -> cg.engine.ProcessorSampler:
         """Returns the default sampler backed by the engine.
@@ -131,14 +131,14 @@ def get_sampler(
             else processor.default_device_config_key.config_alias
         )
 
-        if isinstance(device_version, engine_base.Snapshot):
+        if isinstance(device_version, processor_config.Snapshot):
             return processor_sampler.ProcessorSampler(
                 processor=self,
                 snapshot_id=device_version.id,
                 device_config_name=device_config_name,
                 max_concurrent_jobs=max_concurrent_jobs,
             )
-        if isinstance(device_version, engine_base.Run):
+        if isinstance(device_version, processor_config.Run):
             return processor_sampler.ProcessorSampler(
                 processor=self,
                 run_name=device_version.id,
@@ -504,8 +504,8 @@ def get_schedule(
         filter_str = ' AND '.join(filters)
         return self.context.client.list_time_slots(self.project_id, self.processor_id, filter_str)
 
-    def get_config_from_run(
-        self, run_name: str = 'current', config_name: str = 'default'
+    def get_config(
+        self, device_version: processor_config.DeviceVersion | None = None, config_name: str = ''
     ) -> processor_config.ProcessorConfig | None:
         """Retrieves a ProcessorConfig from an automation run.
 
@@ -514,35 +514,20 @@ def get_config_from_run(
 
         Args:
             processor_id: The processor unique identifier.
+            device_version: Specifies either the snapshot_id or the run_name.
             config_name: The quantum processor's unique identifier.
             run_name: The automation run name.  Use 'default'
                       if none id provided.
 
         Returns: The quantum processor config.
         """
-        return self.engine().get_processor_config_from_run(
-            processor_id=self.processor_id, run_name=run_name, config_name=config_name
-        )
-
-    def get_config_from_snapshot(
-        self, snapshot_id: str, config_name: str = 'default'
-    ) -> processor_config.ProcessorConfig | None:
-        """Retrieves a ProcessorConfig from a given snapshot id.
-
-        If not `config_name` is specified, the internally configured default is returned.
-
-        Args:
-            processor_id: The processor unique identifier.
-            config_name: The quantum processor's unique identifier.
-            snapshot_id: The snapshot's unique identifier.
-
-        Returns: The quantum processor config.
-
-        Raises:
-            EngineException: If the request to get the config fails.
-        """
-        return self.engine().get_processor_config_from_snapshot(
-            processor_id=self.processor_id, snapshot_id=snapshot_id, config_name=config_name
+        default_device_key = self._inner_processor().default_device_config_key
+        return self.engine().get_processor_config(
+            processor_id=self.processor_id,
+            device_version=(
+                device_version if device_version else processor_config.Run(default_device_key.run)
+            ),
+            config_name=config_name if config_name else default_device_key.config_alias,
         )
 
     def __str__(self):
diff --git a/cirq-google/cirq_google/engine/engine_processor_test.py b/cirq-google/cirq_google/engine/engine_processor_test.py
index 035a6d8f1e9..5b6d1692b90 100644
--- a/cirq-google/cirq_google/engine/engine_processor_test.py
+++ b/cirq-google/cirq_google/engine/engine_processor_test.py
@@ -29,8 +29,9 @@
 import cirq_google as cg
 from cirq_google.api import v2
 from cirq_google.cloud import quantum
-from cirq_google.engine import engine_client, ProcessorConfig, util
-from cirq_google.engine.engine import EngineContext, Run, Snapshot
+from cirq_google.engine import engine_client, util
+from cirq_google.engine.engine import EngineContext
+from cirq_google.engine.processor_config import ProcessorConfig, Run, Snapshot
 
 
 def _to_timestamp(json_string):
@@ -346,21 +347,21 @@ def test_get_sampler_from_run_name() -> None:
 
 def test_get_sampler_from_run_name_with_defaults() -> None:
     default_config_alias = 'default_alias'
+    default_run_name = 'default_run'
     processor = cg.EngineProcessor(
         'a',
         'p',
         EngineContext(),
         _processor=quantum.QuantumProcessor(
             default_device_config_key=quantum.DeviceConfigKey(
-                run="run", config_alias=default_config_alias
+                run=default_run_name, config_alias=default_config_alias
             )
         ),
     )
-    run = Run(id='test_run_name')
 
-    sampler = processor.get_sampler(device_version=run)
+    sampler = processor.get_sampler()
 
-    assert sampler.run_name == run.id
+    assert sampler.run_name == default_run_name
     assert sampler.device_config_name == default_config_alias
 
 
@@ -1033,12 +1034,13 @@ def test_str():
 def test_get_config_from_run(client):
     project_id = "test_project_id"
     processor_id = "test_proc_id"
-    run_name = "test_run_name"
+    run = Run(id="test_run_name")
     config_name = "test_config_name"
+    test_snapshot = "test_snapshot"
     name = (
         f'projects/{project_id}/'
         f'processors/{processor_id}/'
-        f'configAutomationRuns/{run_name}/'
+        f'configSnapshots/{test_snapshot}/'
         f'configs/{config_name}'
     )
 
@@ -1057,26 +1059,32 @@ def test_get_config_from_run(client):
         device_specification=util.pack_any(device_spec),
         characterization=util.pack_any(_METRIC_SNAPSHOT),
     )
-    client().get_quantum_processor_config_from_run_async.return_value = quantum_config
-    expected_config = ProcessorConfig(quantum_processor_config=quantum_config, run_name=run_name)
     processor = cg.EngineProcessor(
         project_id=project_id, processor_id=processor_id, context=EngineContext()
     )
 
-    actual_config = processor.get_config_from_run(config_name=config_name, run_name=run_name)
+    client().get_quantum_processor_config_async.return_value = quantum_config
+    expected_config = ProcessorConfig(
+        processor=processor, quantum_processor_config=quantum_config, device_version=run
+    )
+
+    actual_config = processor.get_config(config_name=config_name, device_version=run)
 
-    client().get_quantum_processor_config_from_run_async.assert_called_once_with(
-        project_id=project_id, processor_id=processor_id, run_name=run_name, config_name=config_name
+    client().get_quantum_processor_config_async.assert_called_once_with(
+        project_id=project_id,
+        processor_id=processor_id,
+        device_version=run,
+        config_name=config_name,
     )
     assert actual_config.processor_id == expected_config.processor_id
     assert actual_config.config_name == config_name
-    assert actual_config.run_name == run_name
+    assert actual_config.run_name == run.id
     assert actual_config.effective_device == expected_config.effective_device
     assert actual_config.calibration == expected_config.calibration
 
 
 @mock.patch('cirq_google.engine.engine_client.EngineClient', autospec=True)
-def test_get_default_config_from_run(client):
+def test_get_default_config(client):
     project_id = "test_project_id"
     processor_id = "test_proc_id"
     name = (
@@ -1097,18 +1105,31 @@ def test_get_default_config_from_run(client):
     )
     quantum_config = quantum.QuantumProcessorConfig(
         name=name,
-        device_specification=util.pack_any(device_spec),
+        device_specification=util.pack_any(_DEVICE_SPEC),
         characterization=util.pack_any(_METRIC_SNAPSHOT),
     )
-    client().get_quantum_processor_config_from_run_async.return_value = quantum_config
+    client().get_quantum_processor_config_async.return_value = quantum_config
+
+    default_run = 'current'
+    default_config = 'config_alias'
     processor = cg.EngineProcessor(
-        project_id=project_id, processor_id=processor_id, context=EngineContext()
+        project_id=project_id,
+        processor_id=processor_id,
+        context=EngineContext(),
+        _processor=quantum.QuantumProcessor(
+            default_device_config_key=quantum.DeviceConfigKey(
+                run=default_run, config_alias=default_config
+            )
+        ),
     )
 
-    _ = processor.get_config_from_run()
+    _ = processor.get_config()
 
-    client().get_quantum_processor_config_from_run_async.assert_called_once_with(
-        project_id=project_id, processor_id=processor_id, run_name='current', config_name='default'
+    client().get_quantum_processor_config_async.assert_called_once_with(
+        project_id=project_id,
+        processor_id=processor_id,
+        device_version=Run(id=default_run),
+        config_name=default_config,
     )
 
 
@@ -1116,12 +1137,12 @@ def test_get_default_config_from_run(client):
 def test_get_config_from_snapshot(client):
     project_id = "test_project_id"
     processor_id = "test_proc_id"
-    snapshot_id = "test_snapshot_id"
+    snapshot = Snapshot(id="test_snapshot_id")
     config_name = "test_config_name"
     name = (
         f'projects/{project_id}/'
         f'processors/{processor_id}/'
-        f'configSnapshots/{snapshot_id}/'
+        f'configSnapshots/{snapshot.id}/'
         f'configs/{config_name}'
     )
 
@@ -1140,120 +1161,51 @@ def test_get_config_from_snapshot(client):
         device_specification=util.pack_any(device_spec),
         characterization=util.pack_any(_METRIC_SNAPSHOT),
     )
-    client().get_quantum_processor_config_from_snapshot_async.return_value = quantum_config
-    expected_config = ProcessorConfig(quantum_processor_config=quantum_config)
     processor = cg.EngineProcessor(
         project_id=project_id, processor_id=processor_id, context=EngineContext()
     )
 
-    actual_config = processor.get_config_from_snapshot(
-        config_name=config_name, snapshot_id=snapshot_id
+    client().get_quantum_processor_config_async.return_value = quantum_config
+    expected_config = ProcessorConfig(
+        processor=processor, quantum_processor_config=quantum_config, device_version=snapshot
     )
 
-    client().get_quantum_processor_config_from_snapshot_async.assert_called_once_with(
+    actual_config = processor.get_config(config_name=config_name, device_version=snapshot)
+
+    client().get_quantum_processor_config_async.assert_called_once_with(
         project_id=project_id,
         processor_id=processor_id,
-        snapshot_id=snapshot_id,
+        device_version=snapshot,
         config_name=config_name,
     )
     assert actual_config.processor_id == expected_config.processor_id
     assert actual_config.config_name == config_name
     assert actual_config.run_name == ''
-    assert actual_config.snapshot_id == snapshot_id
+    assert actual_config.snapshot_id == snapshot.id
     assert actual_config.effective_device == expected_config.effective_device
     assert actual_config.calibration == expected_config.calibration
 
 
 @mock.patch('cirq_google.engine.engine_client.EngineClient', autospec=True)
-def test_get_default_config_from_snapshot(client):
+def test_get_config_not_found(client):
     project_id = "test_project_id"
     processor_id = "test_proc_id"
-    snapshot_id = "test_snapshot_id"
-    name = (
-        f'projects/{project_id}/'
-        f'processors/{processor_id}/'
-        f'configSnapshots/{snapshot_id}/'
-        f'configs/default'
-    )
+    default_run = Run(id="current")
+    config_name = "default"
 
-    device_spec = v2.device_pb2.DeviceSpecification(
-        valid_qubits=["0_0", "1_1", "2_2"],
-        valid_targets=[
-            v2.device_pb2.TargetSet(
-                name="2_quibit_targets",
-                target_ordering=v2.device_pb2.TargetSet.SYMMETRIC,
-                targets=[v2.device_pb2.Target(ids=["0_0", "1_1"])],
-            )
-        ],
-    )
-    quantum_config = quantum.QuantumProcessorConfig(
-        name=name,
-        device_specification=util.pack_any(device_spec),
-        characterization=util.pack_any(_METRIC_SNAPSHOT),
-    )
-    client().get_quantum_processor_config_from_snapshot_async.return_value = quantum_config
-    expected_config = ProcessorConfig(quantum_processor_config=quantum_config)
-    processor = cg.EngineProcessor(
-        project_id=project_id, processor_id=processor_id, context=EngineContext()
-    )
-
-    actual_config = processor.get_config_from_snapshot(snapshot_id=snapshot_id)
-
-    client().get_quantum_processor_config_from_snapshot_async.assert_called_once_with(
-        project_id=project_id,
-        processor_id=processor_id,
-        snapshot_id=snapshot_id,
-        config_name='default',
-    )
-    assert actual_config.processor_id == expected_config.processor_id
-    assert actual_config.config_name == 'default'
-    assert actual_config.run_name == ''
-    assert actual_config.snapshot_id == snapshot_id
-    assert actual_config.effective_device == expected_config.effective_device
-    assert actual_config.calibration == expected_config.calibration
-
-
-@mock.patch('cirq_google.engine.engine_client.EngineClient', autospec=True)
-def test_get_config_from_snapshot_not_found(client):
-    project_id = "test_project_id"
-    processor_id = "test_proc_id"
-    snapshot_id = "test_snapshot_id"
-    config_name = "test_config_name"
-
-    client().get_quantum_processor_config_from_snapshot_async.return_value = None
+    client().get_quantum_processor_config_async.return_value = None
 
     processor = cg.EngineProcessor(
         project_id=project_id, processor_id=processor_id, context=EngineContext()
     )
 
-    result = processor.get_config_from_snapshot(config_name=config_name, snapshot_id=snapshot_id)
+    result = processor.get_config(default_run, config_name=config_name)
 
-    client().get_quantum_processor_config_from_snapshot_async.assert_called_once_with(
+    client().get_quantum_processor_config_async.assert_called_once_with(
         project_id=project_id,
         processor_id=processor_id,
-        snapshot_id=snapshot_id,
+        device_version=default_run,
         config_name=config_name,
     )
 
     assert result is None
-
-
-@mock.patch('cirq_google.engine.engine_client.EngineClient', autospec=True)
-def test_get_current_config_from_run_not_found(client):
-    project_id = "test_project_id"
-    processor_id = "test_proc_id"
-    config_name = "test_config_name"
-    run_name = 'test_run_name'
-
-    client().get_quantum_processor_config_from_run_async.return_value = None
-
-    processor = cg.EngineProcessor(
-        project_id=project_id, processor_id=processor_id, context=EngineContext()
-    )
-
-    result = processor.get_config_from_run(config_name=config_name, run_name=run_name)
-
-    client().get_quantum_processor_config_from_run_async.assert_called_once_with(
-        project_id=project_id, processor_id=processor_id, run_name=run_name, config_name=config_name
-    )
-    assert result is None
diff --git a/cirq-google/cirq_google/engine/engine_test.py b/cirq-google/cirq_google/engine/engine_test.py
index 2d1474bf329..f4ae1e88bc0 100644
--- a/cirq-google/cirq_google/engine/engine_test.py
+++ b/cirq-google/cirq_google/engine/engine_test.py
@@ -32,7 +32,8 @@
 from cirq_google.api import v1, v2
 from cirq_google.cloud import quantum
 from cirq_google.engine import util
-from cirq_google.engine.engine import EngineContext, Run, Snapshot
+from cirq_google.engine.engine import EngineContext
+from cirq_google.engine.processor_config import Run, Snapshot
 
 _CIRCUIT = cirq.Circuit(
     cirq.X(cirq.GridQubit(5, 2)) ** 0.5, cirq.measure(cirq.GridQubit(5, 2), key='result')
@@ -953,49 +954,45 @@ def test_get_engine_device(get_processor):
         device.validate_operation(cirq.CZ(cirq.GridQubit(1, 1), cirq.GridQubit(2, 2)))
 
 
-@mock.patch(
-    'cirq_google.engine.engine_client.EngineClient.get_quantum_processor_config_from_snapshot_async'
-)
+@mock.patch('cirq_google.engine.engine_client.EngineClient.get_quantum_processor_config_async')
 def test_get_processor_config_from_snapshot(get_quantum_config_async):
     project_id = "test_project_id"
     processor_id = "test_processor_id"
-    snapshot_id = "test_snapshot_id"
+    snapshot = Snapshot(id="test_snapshot_id")
     config_name = "test_config_name"
     resource_name = (
         f'projects/{project_id}/'
         f'processors/{processor_id}/'
-        f'configSnapshots/{snapshot_id}/'
+        f'configSnapshots/{snapshot.id}/'
         f'configs/{config_name}'
     )
     quantum_confg = quantum.QuantumProcessorConfig(name=resource_name)
 
     get_quantum_config_async.return_value = quantum_confg
 
-    result = cg.Engine(project_id=project_id).get_processor_config_from_snapshot(
-        processor_id=processor_id, snapshot_id=snapshot_id, config_name=config_name
+    result = cg.Engine(project_id=project_id).get_processor_config(
+        processor_id=processor_id, device_version=snapshot, config_name=config_name
     )
 
     get_quantum_config_async.assert_called_with(
         project_id=project_id,
         processor_id=processor_id,
-        snapshot_id=snapshot_id,
+        device_version=snapshot,
         config_name=config_name,
     )
     assert result.processor_id == processor_id
-    assert result.snapshot_id == snapshot_id
+    assert result.snapshot_id == snapshot.id
     assert result.config_name == config_name
     assert result.run_name == ''
 
 
-@mock.patch(
-    'cirq_google.engine.engine_client.EngineClient.get_quantum_processor_config_from_run_async'
-)
+@mock.patch('cirq_google.engine.engine_client.EngineClient.get_quantum_processor_config_async')
 def test_get_processor_config_from_run(get_quantum_config_async):
     project_id = "test_project_id"
     processor_id = "test_processor_id"
     snapshot_id = "test_snapshot_id"
     config_name = "test_config_name"
-    run_name = "test_run_name"
+    run = Run(id="test_run_name")
     resource_name = (
         f'projects/{project_id}/'
         f'processors/{processor_id}/'
@@ -1006,50 +1003,32 @@ def test_get_processor_config_from_run(get_quantum_config_async):
 
     get_quantum_config_async.return_value = quantum_confg
 
-    result = cg.Engine(project_id=project_id).get_processor_config_from_run(
-        processor_id=processor_id, run_name=run_name, config_name=config_name
+    result = cg.Engine(project_id=project_id).get_processor_config(
+        processor_id=processor_id, device_version=run, config_name=config_name
     )
 
     get_quantum_config_async.assert_called_with(
-        project_id=project_id, processor_id=processor_id, run_name=run_name, config_name=config_name
+        project_id=project_id,
+        processor_id=processor_id,
+        device_version=run,
+        config_name=config_name,
     )
     assert result.processor_id == processor_id
     assert result.snapshot_id == snapshot_id
-    assert result.run_name == run_name
+    assert result.run_name == run.id
     assert result.config_name == config_name
 
 
-@mock.patch(
-    'cirq_google.engine.engine_client.EngineClient.get_quantum_processor_config_from_snapshot_async'
-)
+@mock.patch('cirq_google.engine.engine_client.EngineClient.get_quantum_processor_config_async')
 def test_get_processor_config_from_snapshot_none(get_quantum_config_async):
-    project_id = "test_project_id"
-    processor_id = "test_processor_id"
-    snapshot_id = "test_snapshot_id"
-    config_name = "test_config_name"
-
-    get_quantum_config_async.return_value = None
-
-    result = cg.Engine(project_id=project_id).get_processor_config_from_snapshot(
-        processor_id=processor_id, snapshot_id=snapshot_id, config_name=config_name
-    )
-
-    assert result is None
-
-
-@mock.patch(
-    'cirq_google.engine.engine_client.EngineClient.get_quantum_processor_config_from_run_async'
-)
-def test_get_processor_config_from_run_nine(get_quantum_config_async):
     project_id = "test_project_id"
     processor_id = "test_processor_id"
     config_name = "test_config_name"
-    run_name = "test_run_name"
 
     get_quantum_config_async.return_value = None
 
-    result = cg.Engine(project_id=project_id).get_processor_config_from_run(
-        processor_id=processor_id, run_name=run_name, config_name=config_name
+    result = cg.Engine(project_id=project_id).get_processor_config(
+        processor_id=processor_id, config_name=config_name
     )
 
     assert result is None
diff --git a/cirq-google/cirq_google/engine/processor_config.py b/cirq-google/cirq_google/engine/processor_config.py
index a336ce55824..dca38597208 100644
--- a/cirq-google/cirq_google/engine/processor_config.py
+++ b/cirq-google/cirq_google/engine/processor_config.py
@@ -14,17 +14,31 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, TypeAlias
 
 import cirq_google as cg
 from cirq_google.api import v2
-from cirq_google.engine import util
+from cirq_google.engine import processor_sampler, util
 
 if TYPE_CHECKING:
     import cirq
     from cirq_google.cloud.quantum_v1alpha1.types import quantum
 
 
+@dataclass
+class Snapshot:
+    id: str
+
+
+@dataclass
+class Run:
+    id: str
+
+
+DeviceVersion: TypeAlias = Snapshot | Run
+
+
 class ProcessorConfig:
     """Representation of a quantum processor configuration.
 
@@ -33,15 +47,19 @@ class ProcessorConfig:
     """
 
     def __init__(
-        self, *, quantum_processor_config: quantum.QuantumProcessorConfig, run_name: str = ''
+        self,
+        *,
+        quantum_processor_config: quantum.QuantumProcessorConfig,
+        processor: cg.engine.AbstractProcessor,
+        device_version: DeviceVersion | None = None,
     ) -> None:
         """Contructs a Processor Config.
 
         Args:
             quantum_processor_config: The quantum processor config.
+            processor: The processor that this config describes.
         """
         self._quantum_processor_config = quantum_processor_config
-        self._run_name = run_name
         self._grid_device = cg.GridDevice.from_proto(
             util.unpack_any(
                 self._quantum_processor_config.device_specification,
@@ -53,6 +71,8 @@ def __init__(
                 self._quantum_processor_config.characterization, v2.metrics_pb2.MetricsSnapshot()
             )
         )
+        self._device_vesion = device_version
+        self._processor = processor
 
     @property
     def effective_device(self) -> cirq.Device:
@@ -78,7 +98,7 @@ def snapshot_id(self) -> str:
     @property
     def run_name(self) -> str:
         """The run that generated this config if avaiable."""
-        return self._run_name
+        return self._device_vesion.id if isinstance(self._device_vesion, Run) else ''
 
     @property
     def processor_id(self) -> str:
@@ -92,6 +112,28 @@ def config_name(self) -> str:
         parts = self._quantum_processor_config.name.split('/')
         return parts[-1]
 
+    def sampler(self, max_concurrent_jobs: int = 100) -> processor_sampler.ProcessorSampler:
+        """Returns the sampler backed by this config.
+
+        Args:
+            max_concurrent_jobs: The maximum number of jobs to be sent
+                simultaneously to the Engine. This client-side throttle can be
+                used to proactively reduce load to the backends and avoid quota
+                violations when pipelining circuit executions.
+
+        Returns:
+            A `cirq.Sampler` instance (specifically a `engine_sampler.ProcessorSampler`
+            that will send circuits to the Quantum Computing Service
+            when sampled.
+        """
+        return processor_sampler.ProcessorSampler(
+            processor=self._processor,
+            run_name=self.run_name,
+            snapshot_id=self.snapshot_id,
+            device_config_name=self.config_name,
+            max_concurrent_jobs=max_concurrent_jobs,
+        )
+
     def __repr__(self) -> str:
         return (
             'cirq_google.ProcessorConfig'
diff --git a/cirq-google/cirq_google/engine/processor_config_test.py b/cirq-google/cirq_google/engine/processor_config_test.py
index fcba9a4ef58..4436c35d563 100644
--- a/cirq-google/cirq_google/engine/processor_config_test.py
+++ b/cirq-google/cirq_google/engine/processor_config_test.py
@@ -21,6 +21,7 @@
 from cirq_google.cloud import quantum
 from cirq_google.devices import GridDevice
 from cirq_google.engine import util
+from cirq_google.engine.processor_config import Run, Snapshot
 
 _METRIC_SNAPSHOT = v2.metrics_pb2.MetricsSnapshot(
     timestamp_ms=1562544000021,
@@ -58,7 +59,9 @@
 
 
 def test_processor_config_snapshot_id():
-    config = cg.engine.ProcessorConfig(quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG)
+    config = cg.engine.ProcessorConfig(
+        processor=None, quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG
+    )
 
     assert config.snapshot_id == _SNAPSHOT_ID
 
@@ -69,46 +72,56 @@ def test_processor_config_snapshot_id_empty():
         device_specification=util.pack_any(_DEVICE_SPEC),
         characterization=util.pack_any(_METRIC_SNAPSHOT),
     )
-    config = cg.engine.ProcessorConfig(quantum_processor_config=quantum_config)
+    config = cg.engine.ProcessorConfig(processor=None, quantum_processor_config=quantum_config)
 
     assert config.snapshot_id == ''
 
 
 def test_processor_config_run_name():
-    run_name = 'test_run_name'
+    run = Run(id='test_run_name')
     config = cg.engine.ProcessorConfig(
-        quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG, run_name=run_name
+        processor=None, quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG, device_version=run
     )
 
-    assert config.run_name == run_name
+    assert config.run_name == run.id
 
 
 def test_processor_config_effective_device():
-    config = cg.engine.ProcessorConfig(quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG)
+    config = cg.engine.ProcessorConfig(
+        processor=None, quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG
+    )
 
     assert config.effective_device == GridDevice.from_proto(_DEVICE_SPEC)
 
 
 def test_processor_config_calibration():
-    config = cg.engine.ProcessorConfig(quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG)
+    config = cg.engine.ProcessorConfig(
+        processor=None, quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG
+    )
 
     assert config.calibration == cg.Calibration(_METRIC_SNAPSHOT)
 
 
 def test_processor_processor_id():
-    config = cg.engine.ProcessorConfig(quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG)
+    config = cg.engine.ProcessorConfig(
+        processor=None, quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG
+    )
 
     assert config.processor_id == _PROCESSOR_ID
 
 
-def test_processor_CONFIG_NAME():
-    config = cg.engine.ProcessorConfig(quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG)
+def test_processor_config_name():
+    config = cg.engine.ProcessorConfig(
+        processor=None, quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG
+    )
 
     assert config.config_name == _CONFIG_NAME
 
 
 def test_processor_config_repr():
-    config = cg.engine.ProcessorConfig(quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG)
+    config = cg.engine.ProcessorConfig(
+        processor=None, quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG
+    )
     expected_repr = (
         'cirq_google.ProcessorConfig'
         f'processor_id={_PROCESSOR_ID}, '
@@ -121,16 +134,28 @@ def test_processor_config_repr():
 
 
 def test_processor_config_repr_with_run_name():
-    run_name = 'test_run_name'
+    run = Run(id='test_run_name')
     config = cg.engine.ProcessorConfig(
-        quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG, run_name=run_name
+        processor=None, quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG, device_version=run
     )
     expected_repr = (
         'cirq_google.ProcessorConfig'
         f'processor_id={_PROCESSOR_ID}, '
         f'snapshot_id={_SNAPSHOT_ID}, '
-        f'run_name={run_name} '
+        f'run_name={run.id} '
         f'config_name={_CONFIG_NAME}'
     )
 
     assert repr(config) == expected_repr
+
+
+def test_sampler():
+    run = Run(id='test_run_name')
+    config = cg.engine.ProcessorConfig(
+        processor=None, quantum_processor_config=_VALID_QUANTUM_PROCESSOR_CONFIG, device_version=run
+    )
+    sampler = config.sampler()
+
+    assert sampler.run_name == run.id
+    assert sampler.snapshot_id == _SNAPSHOT_ID
+    assert sampler.device_config_name == _CONFIG_NAME

From 3ff91b17a0c5b3f6bcf8c1eea8c457692bb4d86c Mon Sep 17 00:00:00 2001
From: dyates <devyates@google.com>
Date: Fri, 24 Oct 2025 20:41:17 +0000
Subject: [PATCH 12/12] Format fixes;Remove unused imports.

---
 cirq-google/cirq_google/engine/engine.py             |  2 +-
 cirq-google/cirq_google/engine/engine_client.py      |  3 ++-
 cirq-google/cirq_google/engine/engine_processor.py   |  2 +-
 .../cirq_google/engine/engine_processor_test.py      | 12 +-----------
 .../cirq_google/engine/processor_config_test.py      |  2 +-
 5 files changed, 6 insertions(+), 15 deletions(-)

diff --git a/cirq-google/cirq_google/engine/engine.py b/cirq-google/cirq_google/engine/engine.py
index 0e33ebb7fe2..493f604c092 100644
--- a/cirq-google/cirq_google/engine/engine.py
+++ b/cirq-google/cirq_google/engine/engine.py
@@ -29,7 +29,7 @@
 import enum
 import random
 import string
-from typing import TYPE_CHECKING, TypeAlias, TypeVar
+from typing import TYPE_CHECKING, TypeVar
 
 import duet
 import google.auth
diff --git a/cirq-google/cirq_google/engine/engine_client.py b/cirq-google/cirq_google/engine/engine_client.py
index de5df2fc3ca..dbabb077391 100644
--- a/cirq-google/cirq_google/engine/engine_client.py
+++ b/cirq-google/cirq_google/engine/engine_client.py
@@ -1283,9 +1283,10 @@ def _quantum_processor_config_name_from_device_version(
             f'configSnapshots/{device_version.id}/'
             f'configs/{config_name}'
         )
+    default_run_name = 'default'
     return (
         f'{processor_resource_name}/'
-        f'configAutomationRuns/{device_version.id if device_version else 'default'}/'
+        f'configAutomationRuns/{device_version.id if device_version else default_run_name}/'
         f'configs/{config_name}'
     )
 
diff --git a/cirq-google/cirq_google/engine/engine_processor.py b/cirq-google/cirq_google/engine/engine_processor.py
index 80395d4cb38..1662e59ecd8 100644
--- a/cirq-google/cirq_google/engine/engine_processor.py
+++ b/cirq-google/cirq_google/engine/engine_processor.py
@@ -17,7 +17,6 @@
 import datetime
 from typing import Any, TYPE_CHECKING
 
-import cirq_google.engine.engine as engine_base
 from cirq import _compat
 from cirq_google.api import v2
 from cirq_google.devices import grid_device
@@ -36,6 +35,7 @@
     import cirq_google as cg
     import cirq_google.cloud.quantum as quantum
     import cirq_google.engine.abstract_job as abstract_job
+    import cirq_google.engine.engine as engine_base
 
 
 def _date_to_timestamp(union_time: datetime.datetime | datetime.date | int | None) -> int | None:
diff --git a/cirq-google/cirq_google/engine/engine_processor_test.py b/cirq-google/cirq_google/engine/engine_processor_test.py
index 5b6d1692b90..dd57b962535 100644
--- a/cirq-google/cirq_google/engine/engine_processor_test.py
+++ b/cirq-google/cirq_google/engine/engine_processor_test.py
@@ -1090,19 +1090,9 @@ def test_get_default_config(client):
     name = (
         f'projects/{project_id}/'
         f'processors/{processor_id}/'
-        f'configAutomationRuns/default/configs/default'
+        'configAutomationRuns/default/configs/default'
     )
 
-    device_spec = v2.device_pb2.DeviceSpecification(
-        valid_qubits=["0_0", "1_1", "2_2"],
-        valid_targets=[
-            v2.device_pb2.TargetSet(
-                name="2_quibit_targets",
-                target_ordering=v2.device_pb2.TargetSet.SYMMETRIC,
-                targets=[v2.device_pb2.Target(ids=["0_0", "1_1"])],
-            )
-        ],
-    )
     quantum_config = quantum.QuantumProcessorConfig(
         name=name,
         device_specification=util.pack_any(_DEVICE_SPEC),
diff --git a/cirq-google/cirq_google/engine/processor_config_test.py b/cirq-google/cirq_google/engine/processor_config_test.py
index 4436c35d563..0ff85078c06 100644
--- a/cirq-google/cirq_google/engine/processor_config_test.py
+++ b/cirq-google/cirq_google/engine/processor_config_test.py
@@ -21,7 +21,7 @@
 from cirq_google.cloud import quantum
 from cirq_google.devices import GridDevice
 from cirq_google.engine import util
-from cirq_google.engine.processor_config import Run, Snapshot
+from cirq_google.engine.processor_config import Run
 
 _METRIC_SNAPSHOT = v2.metrics_pb2.MetricsSnapshot(
     timestamp_ms=1562544000021,