diff --git a/crates/libs/sbx/mgr/server/src/service/create.rs b/crates/libs/sbx/mgr/server/src/service/create.rs
index 8a1a65658..2db754a7f 100644
--- a/crates/libs/sbx/mgr/server/src/service/create.rs
+++ b/crates/libs/sbx/mgr/server/src/service/create.rs
@@ -657,14 +657,18 @@ impl SandboxManagerService {
         let id = &sandbox.id;
         let original_state = sandbox.state;
         let mut uuid8 = sandbox.uuid8.clone();
+        let cache_root = std::path::Path::new(CACHE_DIR);
+        #[cfg(test)]
+        let cache_root = self.create_cache_dir.as_deref().unwrap_or(cache_root);
         let prepare_source = start_stage_timeout(
             "restore.template_prepare_source",
             START_TEMPLATE_RESOLUTION_TIMEOUT,
             id,
-            template_source_for_sandbox_prepare(
+            template_source_for_sandbox_prepare_in(
                 &self.node_manager,
                 self.template_resolver.as_ref(),
                 sandbox,
+                cache_root,
             ),
         )
         .await?;
@@ -709,12 +713,27 @@ impl SandboxManagerService {
             Self::node_sandbox_memory_policy(&mut node_client),
         )
         .await?;
-        let snapshot_dir = format!("{CACHE_DIR}/{id}/snapshot");
+        let snapshot_dir = cache_root
+            .join(id)
+            .join("snapshot")
+            .to_string_lossy()
+            .into_owned();
         let snapshot_execution = checkpoint_execution_metadata(&snapshot_dir)
             .map_err(|error| Status::failed_precondition(error.to_string()))?;
         let cpu_budget = restore_cpu_budget(vcpu, snapshot_execution.cpu_budget())?;
         let metadata_json = activation_metadata_json(sandbox, cpu_budget)?;
-        let requested_execution = snapshot_execution.into_execution();
+        let producer_execution = snapshot_execution.into_execution();
+        let requested_execution = start_stage_timeout(
+            "restore.select_consumer",
+            START_NODE_METADATA_TIMEOUT,
+            id,
+            super::environment_selection::select_with_client(
+                &mut node_client,
+                &producer_execution,
+                pplx_space_node_mgr_client::proto::ExecutionEnvironmentSelection::Restore,
+            ),
+        )
+        .await?;
         let snapshot_memory_mib = ensure_snapshot_memory_allowed(
             snapshot_memory_file_mib(&snapshot_dir)?,
             node_memory_policy.guest_memory_max_mib,
@@ -808,6 +827,16 @@ impl SandboxManagerService {
             },
         )?;
 
+        // Commit the consumer before Prepare can rewrite disk metadata. The
+        // checkpoint manifest remains producer evidence; a cancelled attempt
+        // retains the consumer for a later cold start. Another restore attempt
+        // selects again from the unchanged checkpoint producer.
+        {
+            let db = self.db_conn()?;
+            crate::db::upsert_sandbox_execution_metadata(&db, id, &requested_execution)
+                .map_err(|error| Status::internal(format!("persist restore consumer: {error}")))?;
+        }
+
         // Allocate network slot
         let execution_fence = read_node_execution_fence(&self.db, id)?.ok_or_else(|| {
             Status::failed_precondition("activation has no durable node execution identity")
diff --git a/crates/libs/sbx/mgr/server/src/service/create/tests.rs b/crates/libs/sbx/mgr/server/src/service/create/tests.rs
index 8d024bb05..400ac31c6 100644
--- a/crates/libs/sbx/mgr/server/src/service/create/tests.rs
+++ b/crates/libs/sbx/mgr/server/src/service/create/tests.rs
@@ -98,3 +98,168 @@ async fn create_refusals_before_insert_persist_nothing() {
         );
     }
 }
+
+/// Stop at allocation's validated rollback boundary, before any VM or host
+/// filesystem mutation. Both attempts must select from the same checkpoint.
+#[tokio::test]
+async fn restore_persists_consumer_before_allocation_and_reselects_from_checkpoint() {
+    use crate::node_manager_transport::fake::FakeNodeManager;
+    use pplx_space_common_core::types::{
+        Arch, EnvironmentDescriptor, RuntimeBinding, SandboxArtifactMetadata,
+        SnapshotExecutionMetadata,
+    };
+    use pplx_space_execution_common::environment::EnvironmentDefinition;
+    use pplx_space_node_mgr_client::proto::{
+        AllocateResponse, ExecutionEnvironmentSelection, NetworkPoolStatusResponse,
+        ReleaseResponse, ResolveExecutionEnvironmentRequest, ResolveExecutionEnvironmentResponse,
+        resolve_execution_environment_request::Compatible,
+    };
+    if !crate::service::tests::enter_route_test_process(&[
+        ("EPPO_SDK_KEY", None),
+        ("SPACE_FLAGD_ENABLED", None),
+    ]) {
+        return;
+    }
+    let initial =
+        pplx_space_node_runtime_core::catalog::CompiledEnvironment::for_architecture(Arch::X86_64)
+            .unwrap()
+            .remove(0);
+    let producer = initial.definition().execution_requirements().unwrap();
+    let mut spec = initial.definition().descriptor().spec().clone();
+    let RuntimeBinding::ManagedRelease(release) = &mut spec.runtime;
+    release.runtime.version = "1.15.2".parse().unwrap();
+    let consumer_definition =
+        EnvironmentDefinition::new(EnvironmentDescriptor::try_from(spec).unwrap());
+    let consumer = consumer_definition.execution_requirements().unwrap();
+    let directory = tempfile::tempdir().unwrap();
+    let cache = directory.path().join("cache");
+    let db_path = directory.path().join("sandbox.sqlite");
+    crate::db::create_test_schema_on_disk(&db_path);
+    let conn = crate::db::Connection::open_at(&db_path).unwrap();
+    let id = SandboxId::new();
+    crate::service::recovery::tests::insert_suspended_test_sandbox(&conn, &id.to_string());
+    crate::db::upsert_sandbox_execution_metadata(&conn, &id.to_string(), &producer).unwrap();
+    let instance = cache.join(id.to_string()).join("instance");
+    let snapshot = cache.join(id.to_string()).join("snapshot");
+    std::fs::create_dir_all(&instance).unwrap();
+    std::fs::create_dir_all(&snapshot).unwrap();
+    std::fs::write(instance.join("rootfs.ext4"), []).unwrap();
+    let metadata = SandboxArtifactMetadata::new(
+        id,
+        producer.clone(),
+        "active-template".parse().unwrap(),
+        None,
+    );
+    std::fs::write(
+        instance.parent().unwrap().join("metadata.json"),
+        serde_json::to_vec(&metadata).unwrap(),
+    )
+    .unwrap();
+    let checkpoint =
+        serde_json::to_vec(&SnapshotExecutionMetadata::new(producer.clone()).unwrap()).unwrap();
+    std::fs::write(snapshot.join("runtime.json"), &checkpoint).unwrap();
+    let expected_producer = producer.clone();
+    let expected_consumer = consumer.clone();
+    let selected_consumer = consumer.clone();
+    let observed_db = db_path.clone();
+    let allocations = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+    let observed_allocations = allocations.clone();
+    let node = FakeNodeManager::default()
+        .unary("GetNetworkPoolStatus", |_: NetworkPoolStatusRequest| {
+            Ok(NetworkPoolStatusResponse {
+                sandbox_cgroup_memory_mib: 4096,
+                sandbox_guest_memory_max_mib: 3968,
+                sandbox_memory_headroom_mib: 128,
+                ..Default::default()
+            })
+        })
+        .unary(
+            "ResolveExecutionEnvironment",
+            move |request: ResolveExecutionEnvironmentRequest| {
+                let Some(Compatible::RestoreSource(source)) = request.compatible else {
+                    panic!("restore must use its operation-specific resolver");
+                };
+                assert_eq!(
+                    pplx_space_common_core::types::ExecutionMetadata::try_from(&source).unwrap(),
+                    expected_producer
+                );
+                Ok(ResolveExecutionEnvironmentResponse {
+                    accepted_selection: ExecutionEnvironmentSelection::Restore as i32,
+                    definition: Some((&consumer_definition).into()),
+                    execution: Some((&selected_consumer).into()),
+                })
+            },
+        )
+        .unary("Allocate", move |request: AllocateRequest| {
+            let db = crate::db::Connection::open_at(&observed_db).unwrap();
+            let row = crate::db::get_sandbox(&db, &request.sandbox_id)
+                .unwrap()
+                .unwrap();
+            assert_eq!(row.state, SandboxState::Restoring);
+            assert_eq!(row.execution_metadata, Some(expected_consumer.clone()));
+            observed_allocations.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            Ok(AllocateResponse {
+                sandbox_uid: 1,
+                ..Default::default()
+            })
+        })
+        .unary(
+            "Release",
+            |_: ReleaseRequest| Ok(ReleaseResponse::default()),
+        )
+        .serve();
+    let service = SandboxManagerService::new_with_options(
+        conn,
+        SandboxManagerServiceOptions {
+            node_manager: node.connector(),
+            create_cache_dir: Some(cache),
+            ..SandboxManagerServiceOptions::for_test()
+        },
+    );
+    for attempt in 1..=2 {
+        let row = service.lookup_sandbox(&id.to_string()).unwrap();
+        assert_eq!(row.state, SandboxState::Suspended);
+        let error = tokio::time::timeout(
+            Duration::from_secs(20),
+            service.start_from_checkpoint(StartFromCheckpointLaunch {
+                vcpu: None,
+                activation_auto_lifecycle: None,
+                operation_activation_attempt_id: None,
+                sandbox: &row,
+                blocking: false,
+                netns_path: None,
+                target_state: SandboxState::Started,
+                execution_generation: None,
+                space_api_route_identity: Default::default(),
+                route_identity_provenance:
+                    crate::auto_lifecycle_delete_receipts::RouteIdentityProvenance::Unfenced,
+                open_activation_claim_eligible: false,
+                restored_snapshot: RestoredSnapshotProvenance::default(),
+            }),
+        )
+        .await
+        .unwrap()
+        .unwrap_err();
+        assert!(error.message().contains("invalid sandbox uid"), "{error}");
+        assert_eq!(
+            allocations.load(std::sync::atomic::Ordering::SeqCst),
+            attempt
+        );
+        assert_eq!(
+            service
+                .lookup_sandbox(&id.to_string())
+                .unwrap()
+                .execution_metadata,
+            Some(consumer.clone())
+        );
+        assert_eq!(
+            std::fs::read(snapshot.join("runtime.json")).unwrap(),
+            checkpoint
+        );
+    }
+    service.shutdown_token.cancel();
+    service.task_tracker.close();
+    tokio::time::timeout(Duration::from_secs(20), service.task_tracker.wait())
+        .await
+        .unwrap();
+}
diff --git a/crates/libs/sbx/mgr/server/src/service/mod.rs b/crates/libs/sbx/mgr/server/src/service/mod.rs
index 6be7245b6..d19adebaf 100644
--- a/crates/libs/sbx/mgr/server/src/service/mod.rs
+++ b/crates/libs/sbx/mgr/server/src/service/mod.rs
@@ -1045,7 +1045,7 @@ pub struct SandboxManagerServiceOptions {
     /// reads in test builds. `None` reads the fixed sandbox run directory.
     #[cfg(test)]
     pub(crate) cleanup_run_dir: Option<PathBuf>,
-    /// Cache root used by logical Create in tests; other filesystem paths retain their own fixtures.
+    /// Cache root for logical Create and checkpoint admission tests; later VM work uses its own fixtures.
     #[cfg(test)]
     pub(crate) create_cache_dir: Option<PathBuf>,
 }
diff --git a/docs/execution-environments.md b/docs/execution-environments.md
index 907699b07..551d23edc 100644
--- a/docs/execution-environments.md
+++ b/docs/execution-environments.md
@@ -563,8 +563,8 @@ store and stages its kernel into the execution cache. Cache loss triggers
 reconstruction. Missing or corrupt non-default artifacts cannot replace or
 withdraw E0. Each compiled binding records its qualified lifecycle support; the
 1.17.0 / 6.18.44 and 1.17.0 / 6.1.155 bindings support cold execution and
-same-environment memory snapshots. Restoring across environment identities is
-not an upgrade path: each snapshot retains its original VMM and kernel binding.
+same-environment memory snapshots. Cross-environment restore requires a
+qualified compatible loader; each snapshot retains its original producer binding.
 The 6.1.155 kernel keeps its existing artifact identity when paired with 1.17.0.
 Non-default snapshot capture carries an environment-bound sandbox identity on
 `CreateSnapshot`, with the legacy sandbox ID omitted. Old node-managers reject
@@ -765,3 +765,24 @@ infer permission from `SPACE_ENV`: absent settings revoke non-E0 bindings even o
 Kind and development nodes. Before upgrading a deployment that needs those existing
 bindings, configure its explicit allowlist while keeping the default at E0. Do not
 expand every development deployment's policy merely because of its name.
+
+### Selecting a restore consumer
+
+The snapshot manifest retains original runtime, kernel and state-format identity.
+After claiming activation ownership, sandbox-manager saves the selected loader in
+the sandbox row before preparing files or allocating a slot. Prepare and Launch
+receive that exact consumer. A failed attempt can leave it recorded; cleanup
+retains the checkpoint and a later restore attempt resolves again from producer
+evidence. A subsequent snapshot describes the runtime that actually ran.
+
+Pause/resume keeps the resident VM. This policy change does not kill running VMs
+or add isolated resume refusal: incident-driven stopping, resource reclamation and
+resume revocation must ship together. Stop and delete remain available. Operators
+must stop affected resident VMs when revoking execution during an incident.
+
+A future kernel or incompatible-state upgrade requires explicit memory-loss
+consent, activation fencing and a preserved matching disk/checkpoint pair. Keep
+the old pair until the new cold boot is verified; a failed cold boot must not
+mutate the only rollback disk. Retention is not permission to restore a revoked
+runtime. Retire the checkpoint through a durable lifecycle operation before later
+object garbage collection, never by deleting S3 vmstate and editing lifecycle rows.