diff --git a/crates/libs/build/mgr/src/execution_binding.rs b/crates/libs/build/mgr/src/execution_binding.rs
index e698240d3..1fe878e5f 100644
--- a/crates/libs/build/mgr/src/execution_binding.rs
+++ b/crates/libs/build/mgr/src/execution_binding.rs
@@ -2,9 +2,7 @@
 //! process environment variables and from downloadable artifact catalog entries.
 use std::path::Path;
 
-use pplx_space_common_core::types::{
-    Digest, EnvironmentReference, ExecutionMetadata, TemplateMetadata,
-};
+use pplx_space_common_core::types::{EnvironmentReference, ExecutionMetadata, TemplateMetadata};
 use pplx_space_execution_common::{
     ExecutionProtoError,
     environment::{EnvironmentBindingError, EnvironmentDecodeError, EnvironmentDefinition},
@@ -13,15 +11,6 @@ use pplx_space_node_mgr_client::proto::{
     ResolveExecutionEnvironmentRequest, ResolveExecutionEnvironmentResponse,
 };
 
-// The legacy Create protocol implies the immutable stock E0 binding. This is
-// compatibility evidence, not an approval or selection of another environment.
-const LEGACY_CREATE_ENVIRONMENT: EnvironmentReference =
-    EnvironmentReference::new(Digest::sha256([
-        0x31, 0x68, 0xc9, 0x6b, 0x57, 0x19, 0xce, 0x62, 0x54, 0x3a, 0xe5, 0x3e, 0x66, 0x41, 0xa4,
-        0xeb, 0x80, 0x41, 0x6a, 0x9f, 0x01, 0xe5, 0x58, 0x5f, 0x6f, 0x76, 0xc3, 0xc7, 0x01, 0x11,
-        0xfa, 0x0b,
-    ]));
-
 #[derive(Debug, thiserror::Error)]
 pub enum BuildExecutionBindingError {
     #[error(transparent)]
@@ -42,6 +31,14 @@ pub enum BuildExecutionBindingError {
     Binding(#[from] EnvironmentBindingError),
     #[error("environment resolver returned a different requested binding")]
     Reference,
+    #[error("compatible environment resolution: {0}")]
+    Selection(#[from] pplx_space_node_mgr_client::environment_selection::SelectionResponseError),
+    #[error("scratch build requires Firecracker/Linux/{architecture}, got {runtime:?}/{guest:?}")]
+    ScratchRequirements {
+        architecture: pplx_space_common_core::types::Arch,
+        runtime: pplx_space_common_core::types::RuntimeKind,
+        guest: pplx_space_common_core::types::GuestPlatform,
+    },
     #[error("recipe kernel pin differs from the resolved environment")]
     KernelPin,
     #[error("cannot read template metadata {path}: {source}")]
@@ -67,25 +64,27 @@ impl BuildExecutionBinding {
         self.definition.reference()
     }
 
-    /// Ordinary inherited E0 work remains compatible during reader rollout.
-    /// Every explicit request or other complete binding requires acknowledgement.
-    pub(super) fn create_environment(&self, explicit: bool) -> Option<EnvironmentReference> {
-        (explicit || self.reference() != LEGACY_CREATE_ENVIRONMENT).then(|| self.reference())
+    /// Freeze even E0 before Create; inventory changes must not cause a second
+    /// selection between builder resolution and the sandbox's persisted intent.
+    pub(super) fn create_environment(&self) -> EnvironmentReference {
+        self.reference()
     }
 
-    /// Legacy bases keep their execution contract. Requested and inherited
-    /// references use verified node resolution; wire compatibility for implicit
-    /// E0 remains a separate decision.
     pub(super) async fn for_base(
         reference: Option<EnvironmentReference>,
         execution: &ExecutionMetadata,
     ) -> Result<Option<Self>, BuildExecutionBindingError> {
-        // An explicit request takes precedence, but must match the base evidence.
-        let Some(reference) = reference.or(execution.environment()) else {
+        if let Some(reference) = reference {
+            let binding = Self::resolve(Some(reference)).await?;
+            binding.definition.check_cold_boot_source(execution)?;
+            return Ok(Some(binding));
+        }
+        let Some(reference) = execution.environment() else {
             return Ok(None);
         };
-        let binding = Self::resolve(Some(reference)).await?;
-        binding.bind(execution.clone())?;
+        let request = pplx_space_node_mgr_client::proto::resolve_execution_environment_request::Compatible::ColdBootSource(execution.into());
+        let binding = Self::resolve_compatible(request, reference).await?;
+        binding.definition.check_cold_boot_source(execution)?;
         Ok(Some(binding))
     }
 
@@ -101,7 +100,68 @@ impl BuildExecutionBinding {
         {
             return Ok(None);
         }
-        Self::resolve(reference).await.map(Some)
+        let binding = Self::resolve(reference).await?;
+        binding.check_scratch_requirements(architecture)?;
+        Ok(Some(binding))
+    }
+
+    fn check_scratch_requirements(
+        &self,
+        architecture: pplx_space_common_core::types::Arch,
+    ) -> Result<(), BuildExecutionBindingError> {
+        use pplx_space_common_core::types::{GuestOperatingSystem, GuestPlatform, RuntimeKind};
+        let guest = GuestPlatform {
+            os: GuestOperatingSystem::Linux,
+            architecture,
+        };
+        if self.execution.runtime().kind != RuntimeKind::Firecracker
+            || self.execution.guest() != guest
+        {
+            return Err(BuildExecutionBindingError::ScratchRequirements {
+                architecture,
+                runtime: self.execution.runtime().kind,
+                guest: self.execution.guest(),
+            });
+        }
+        Ok(())
+    }
+
+    /// Resolution, including an old-peer exact fallback, shares one deadline.
+    async fn resolve_compatible(
+        request: pplx_space_node_mgr_client::proto::resolve_execution_environment_request::Compatible,
+        fallback_reference: EnvironmentReference,
+    ) -> Result<Self, BuildExecutionBindingError> {
+        use pplx_space_node_mgr_client::environment_selection::{SelectionResponseError, decode};
+        tokio::time::timeout(std::time::Duration::from_secs(30), async {
+            let mut node = pplx_space_node_mgr_client::connect().await?;
+            let response = node
+                .resolve_execution_environment(ResolveExecutionEnvironmentRequest {
+                    reference: None,
+                    compatible: Some(request.clone()),
+                })
+                .await?
+                .into_inner();
+            match decode(&request, response) {
+                Ok(selected) => Ok(Self {
+                    definition: selected.definition,
+                    execution: selected.execution,
+                }),
+                Err(SelectionResponseError::Unsupported) => {
+                    // Mixed rollout retains the pre-existing exact source,
+                    // never an unacknowledged upgrade.
+                    let response = node
+                        .resolve_execution_environment(ResolveExecutionEnvironmentRequest {
+                            reference: Some(fallback_reference.to_string()),
+                            compatible: None,
+                        })
+                        .await?
+                        .into_inner();
+                    Self::from_response(Some(fallback_reference), response)
+                }
+                Err(error) => Err(error.into()),
+            }
+        })
+        .await?
     }
 
     /// Resolve within 30 seconds, including connection setup.
diff --git a/crates/libs/build/mgr/src/execution_binding/tests.rs b/crates/libs/build/mgr/src/execution_binding/tests.rs
index 81eba3b3b..5b8824c5d 100644
--- a/crates/libs/build/mgr/src/execution_binding/tests.rs
+++ b/crates/libs/build/mgr/src/execution_binding/tests.rs
@@ -68,6 +68,18 @@ fn resolution_requires_complete_matching_support_evidence() {
     }
 }
 
+#[test]
+fn scratch_default_must_match_the_builder_guest_requirements() {
+    // Default resolution needs no compatible-selection acknowledgement, including
+    // from an older node. Its verified binding must still fit this builder.
+    let binding = BuildExecutionBinding::from_response(None, fixture()).unwrap();
+    binding.check_scratch_requirements(Arch::X86_64).unwrap();
+    assert!(matches!(
+        binding.check_scratch_requirements(Arch::Aarch64),
+        Err(BuildExecutionBindingError::ScratchRequirements { .. })
+    ));
+}
+
 #[test]
 fn scratch_inputs_replace_inherited_selectors_and_reject_conflicting_recipe_pins() {
     let binding = BuildExecutionBinding::from_response(None, fixture()).unwrap();
@@ -228,19 +240,19 @@ async fn derived_vm_requests_keep_unbound_releases_on_the_legacy_protocol() {
         "seed".into(),
         "sha256:bound-base".into(),
         None,
-        binding.create_environment(false),
+        Some(binding.create_environment()),
         &pplx_space_net_common::NetworkPolicyV1::allow_all(),
     );
-    assert!(
-        inherited_e0.environment.is_none(),
-        "ordinary inherited E0 builds remain rolling-compatible"
+    assert_eq!(
+        inherited_e0.environment.unwrap().reference,
+        binding.reference().to_string()
     );
-    assert_eq!(inherited_e0.runtime, "fc");
+    assert!(inherited_e0.runtime.is_empty());
     let explicit_e0 = super::super::build_vm_create_request(
         "seed".into(),
         "sha256:bound-base".into(),
         None,
-        binding.create_environment(true),
+        Some(binding.create_environment()),
         &pplx_space_net_common::NetworkPolicyV1::allow_all(),
     );
     assert_eq!(
@@ -264,8 +276,5 @@ async fn derived_vm_requests_keep_unbound_releases_on_the_legacy_protocol() {
         },
     )
     .unwrap();
-    assert_eq!(
-        other.create_environment(false),
-        Some(definition.reference())
-    );
+    assert_eq!(other.create_environment(), definition.reference());
 }
diff --git a/crates/libs/build/mgr/src/lib.rs b/crates/libs/build/mgr/src/lib.rs
index a864cb8fa..95dcf340a 100644
--- a/crates/libs/build/mgr/src/lib.rs
+++ b/crates/libs/build/mgr/src/lib.rs
@@ -806,7 +806,7 @@ impl TemplateBuilder {
                 Some(build_rootfs_size_mib),
                 execution_binding
                     .as_ref()
-                    .and_then(|binding| binding.create_environment(self.environment.is_some())),
+                    .map(BuildExecutionBinding::create_environment),
                 &self.network_policy,
             );
             Ok::<String, BuildError>(
diff --git a/crates/libs/sbx/mgr/server/src/service/create_support.rs b/crates/libs/sbx/mgr/server/src/service/create_support.rs
index 6b94a0357..a85b1f200 100644
--- a/crates/libs/sbx/mgr/server/src/service/create_support.rs
+++ b/crates/libs/sbx/mgr/server/src/service/create_support.rs
@@ -932,44 +932,58 @@ pub(super) async fn template_source_for_sandbox_prepare_in(
     Ok(SandboxPrepareSource {
         template: Some(sandbox.template.clone()),
         template_dir: Some(resolved_template.template_dir),
-        execution: bind_saved_execution(sandbox, resolved_template.execution)?,
+        execution: bind_disk_execution(
+            sandbox.execution_metadata.as_ref(),
+            resolved_template.execution,
+        )?,
         existing_artifact: false,
     })
 }
 
-/// Saved selection wins over absent legacy annotations, never over conflicting
-/// artifact evidence. Kernel directory labels do not change byte identity.
+/// The saved sandbox binding owns execution selection. Disk metadata records
+/// the runtime that last prepared the disk; it is not a second selection pin.
+/// Raw disks can cross qualified consumers while their guest and kernel inputs
+/// remain fixed. Snapshot producer evidence is read separately from its manifest.
 fn bind_saved_execution(
     sandbox: &crate::db::SandboxRow,
     observed: pplx_space_common_core::types::ExecutionMetadata,
 ) -> Result<pplx_space_common_core::types::ExecutionMetadata, Status> {
-    let Some((saved, reference)) = sandbox
-        .execution_metadata
-        .as_ref()
-        .and_then(|saved| saved.environment().map(|reference| (saved, reference)))
-    else {
+    bind_disk_execution(sandbox.execution_metadata.as_ref(), observed)
+}
+
+fn bind_disk_execution(
+    saved: Option<&pplx_space_common_core::types::ExecutionMetadata>,
+    observed: pplx_space_common_core::types::ExecutionMetadata,
+) -> Result<pplx_space_common_core::types::ExecutionMetadata, Status> {
+    let Some(saved) = saved.filter(|saved| saved.environment().is_some()) else {
         return Ok(observed);
     };
-    if observed
-        .environment()
-        .is_some_and(|actual| actual != reference)
+    // Only the Firecracker adapter currently declares cross-consumer raw-disk
+    // compatibility. Other adapters must supply their own disk contract.
+    if saved.runtime().kind != pplx_space_common_core::types::RuntimeKind::Firecracker
+        && saved != &observed
     {
         return Err(Status::failed_precondition(
-            "artifact environment differs from the sandbox's saved binding",
+            "runtime adapter requires the exact disk execution contract",
         ));
     }
-    if observed.runtime() != saved.runtime() || observed.guest() != saved.guest() {
-        return Err(Status::failed_precondition(
-            "artifact runtime or guest differs from the sandbox's saved binding",
-        ));
-    }
-    if observed.state_format() != saved.state_format() {
+    if saved.runtime().kind != observed.runtime().kind
+        || saved.guest() != observed.guest()
+        || !matching_execution_components(saved, &observed)
+    {
         return Err(Status::failed_precondition(
-            "artifact state format differs from the sandbox's saved binding",
+            "disk boot inputs differ from the sandbox's saved execution",
         ));
     }
-    if observed.components().len() != saved.components().len()
-        || !saved.components().iter().all(|expected| {
+    Ok(saved.clone())
+}
+
+fn matching_execution_components(
+    saved: &pplx_space_common_core::types::ExecutionMetadata,
+    observed: &pplx_space_common_core::types::ExecutionMetadata,
+) -> bool {
+    observed.components().len() == saved.components().len()
+        && saved.components().iter().all(|expected| {
             observed.components().iter().any(|actual| {
                 expected.kind == actual.kind
                     && expected.role == actual.role
@@ -983,12 +997,6 @@ fn bind_saved_execution(
                     }
             })
         })
-    {
-        return Err(Status::failed_precondition(
-            "artifact components differ from the sandbox's saved binding",
-        ));
-    }
-    Ok(observed.with_environment(reference))
 }
 
 pub(super) fn guest_console_logs_enabled() -> bool {
diff --git a/crates/libs/sbx/mgr/server/src/service/create_support/tests.rs b/crates/libs/sbx/mgr/server/src/service/create_support/tests.rs
index f0252c395..647d1bc13 100644
--- a/crates/libs/sbx/mgr/server/src/service/create_support/tests.rs
+++ b/crates/libs/sbx/mgr/server/src/service/create_support/tests.rs
@@ -742,3 +742,35 @@ async fn retained_instance_permission_recovery_and_repair_preserve_disk() {
         }
     }
 }
+
+#[test]
+fn template_provenance_does_not_overwrite_a_selected_cold_boot_consumer() {
+    use pplx_space_common_core::types::Arch;
+    use pplx_space_node_runtime_core::catalog::CompiledEnvironment;
+    let catalog = CompiledEnvironment::for_architecture(Arch::X86_64).unwrap();
+    let source = catalog[0].definition().execution_requirements().unwrap();
+    let selected = catalog
+        .iter()
+        .find(|entry| {
+            entry.release().release.version.as_str() == "1.17.0"
+                && entry.kernel().version.as_str() == "6.1.155"
+        })
+        .unwrap()
+        .definition()
+        .execution_requirements()
+        .unwrap();
+    assert_ne!(source.runtime(), selected.runtime());
+    assert_eq!(
+        bind_disk_execution(Some(&selected), source.clone()).unwrap(),
+        selected
+    );
+    let other_kernel = catalog
+        .iter()
+        .find(|entry| entry.kernel().version.as_str() == "6.18.44")
+        .unwrap()
+        .definition()
+        .execution_requirements()
+        .unwrap();
+    bind_disk_execution(Some(&selected), other_kernel).unwrap_err();
+    assert_eq!(bind_disk_execution(None, source.clone()).unwrap(), source);
+}
diff --git a/crates/libs/sbx/mgr/server/src/service/environment_selection.rs b/crates/libs/sbx/mgr/server/src/service/environment_selection.rs
new file mode 100644
index 000000000..ac508f087
--- /dev/null
+++ b/crates/libs/sbx/mgr/server/src/service/environment_selection.rs
@@ -0,0 +1,119 @@
+//! Cold creates persist selection before activation. Restore attempts select
+//! from immutable checkpoint provenance; retained cold starts use the saved consumer.
+use super::{SandboxManagerService, tail_support::connect_node};
+use pplx_space_common_core::types::{EnvironmentReference, ExecutionMetadata};
+use pplx_space_node_mgr_client::{
+    NodeManagerGrpcClient,
+    environment_selection::{SelectedEnvironment, SelectionResponseError, decode},
+    proto::{
+        ExecutionEnvironmentSelection, ResolveExecutionEnvironmentRequest,
+        resolve_execution_environment_request::Compatible,
+    },
+};
+use tonic::Status;
+
+impl SandboxManagerService {
+    pub(super) async fn select_compatible_execution(
+        &self,
+        source: &ExecutionMetadata,
+        intent: ExecutionEnvironmentSelection,
+    ) -> Result<ExecutionMetadata, Status> {
+        if source.environment().is_none() {
+            return Ok(source.clone());
+        }
+        tokio::time::timeout(std::time::Duration::from_secs(30), async {
+            let mut node = connect_node(&self.node_manager)
+                .await
+                .map_err(|error| Status::unavailable(error.to_string()))?;
+            select_with_client(&mut node, source, intent).await
+        })
+        .await
+        .map_err(|_| Status::deadline_exceeded("compatible environment selection timed out"))?
+    }
+}
+
+/// The caller owns the deadline, including any exact fallback on this connection.
+pub(super) async fn select_with_client(
+    node: &mut NodeManagerGrpcClient,
+    source: &ExecutionMetadata,
+    intent: ExecutionEnvironmentSelection,
+) -> Result<ExecutionMetadata, Status> {
+    let Some(reference) = source.environment() else {
+        return Ok(source.clone());
+    };
+    let request = match intent {
+        ExecutionEnvironmentSelection::ColdBoot => Compatible::ColdBootSource(source.into()),
+        ExecutionEnvironmentSelection::Restore => Compatible::RestoreSource(source.into()),
+        _ => {
+            return Err(Status::invalid_argument(
+                "sandbox selection requires a source operation",
+            ));
+        }
+    };
+    let response = node
+        .resolve_execution_environment(ResolveExecutionEnvironmentRequest {
+            reference: None,
+            compatible: Some(request.clone()),
+        })
+        .await?
+        .into_inner();
+    match decode(&request, response) {
+        Ok(selected) => Ok(selected.execution),
+        Err(SelectionResponseError::Unsupported)
+            if intent == ExecutionEnvironmentSelection::ColdBoot =>
+        {
+            let selected = resolve_exact(node, reference).await?;
+            selected
+                .definition
+                .check_execution(source)
+                .map_err(|error| Status::failed_precondition(error.to_string()))?;
+            Ok(selected.execution)
+        }
+        // Restore admission remains operation-specific in Prepare/Launch. The
+        // exact resolver is a cold-boot RPC and would require unused boot files.
+        Err(SelectionResponseError::Unsupported) => Ok(source.clone()),
+        Err(error) => Err(Status::failed_precondition(error.to_string())),
+    }
+}
+
+pub(super) async fn resolve_exact(
+    node: &mut NodeManagerGrpcClient,
+    reference: EnvironmentReference,
+) -> Result<SelectedEnvironment, Status> {
+    let response = node
+        .resolve_execution_environment(ResolveExecutionEnvironmentRequest {
+            compatible: None,
+            reference: Some(reference.to_string()),
+        })
+        .await?
+        .into_inner();
+    let definition = pplx_space_execution_common::environment::EnvironmentDefinition::try_from(
+        response.definition.as_ref().ok_or_else(|| {
+            Status::failed_precondition("node did not return environment definition")
+        })?,
+    )
+    .map_err(|error| Status::failed_precondition(error.to_string()))?;
+    let actual = pplx_space_common_core::types::ExecutionMetadata::try_from(
+        response.execution.as_ref().ok_or_else(|| {
+            Status::failed_precondition("node did not acknowledge execution environment")
+        })?,
+    )
+    .map_err(|error| Status::failed_precondition(error.to_string()))?;
+    if definition.reference() != reference || actual.environment() != Some(reference) {
+        return Err(Status::failed_precondition(
+            "node returned a different or missing environment binding",
+        ));
+    }
+    definition
+        .check_execution(&actual)
+        .map_err(|error| Status::failed_precondition(error.to_string()))?;
+    Ok(
+        pplx_space_node_mgr_client::environment_selection::SelectedEnvironment {
+            definition,
+            execution: actual,
+        },
+    )
+}
+
+#[cfg(test)]
+mod tests;
diff --git a/crates/libs/sbx/mgr/server/src/service/environment_selection/tests.rs b/crates/libs/sbx/mgr/server/src/service/environment_selection/tests.rs
new file mode 100644
index 000000000..8563a0842
--- /dev/null
+++ b/crates/libs/sbx/mgr/server/src/service/environment_selection/tests.rs
@@ -0,0 +1,110 @@
+use super::*;
+use crate::node_manager_transport::fake::FakeNodeManager;
+use crate::service::SandboxManagerServiceOptions;
+use pplx_space_common_core::types::Arch;
+use pplx_space_node_mgr_client::proto::ResolveExecutionEnvironmentResponse;
+use pplx_space_node_runtime_core::catalog::CompiledEnvironment;
+
+#[tokio::test]
+async fn only_missing_acknowledgement_retains_the_exact_source_during_rollout() {
+    if !crate::service::tests::enter_route_test_process(&[
+        ("EPPO_SDK_KEY", None),
+        ("SPACE_FLAGD_ENABLED", None),
+    ]) {
+        return;
+    }
+    let catalog = CompiledEnvironment::for_architecture(Arch::X86_64).unwrap();
+    let source = catalog[0].definition().execution_requirements().unwrap();
+    let candidate = catalog
+        .iter()
+        .find(|entry| {
+            entry.release().release.version.as_str() == "1.17.0"
+                && entry.kernel().version.as_str() == "6.1.155"
+        })
+        .unwrap();
+    for (intent, acknowledgement, exact_allowed, expected) in [
+        (
+            ExecutionEnvironmentSelection::ColdBoot,
+            0,
+            true,
+            Some("1.15.1"),
+        ),
+        (ExecutionEnvironmentSelection::ColdBoot, 0, false, None),
+        (
+            ExecutionEnvironmentSelection::Restore,
+            0,
+            false,
+            Some("1.15.1"),
+        ),
+        (
+            ExecutionEnvironmentSelection::ColdBoot,
+            ExecutionEnvironmentSelection::ColdBoot as i32,
+            true,
+            Some("1.17.0"),
+        ),
+        (ExecutionEnvironmentSelection::ColdBoot, 99, true, None),
+    ] {
+        let response = ResolveExecutionEnvironmentResponse {
+            accepted_selection: acknowledgement,
+            definition: Some(candidate.definition().into()),
+            execution: Some((&candidate.definition().execution_requirements().unwrap()).into()),
+        };
+        let exact_definition = catalog[0].definition().clone();
+        let exact_source = source.clone();
+        let node = FakeNodeManager::default()
+            .unary(
+                "ResolveExecutionEnvironment",
+                move |request: ResolveExecutionEnvironmentRequest| {
+                    if let Some(compatible) = request.compatible {
+                        assert!(request.reference.is_none());
+                        assert!(matches!(
+                            (intent, compatible),
+                            (
+                                ExecutionEnvironmentSelection::ColdBoot,
+                                Compatible::ColdBootSource(_)
+                            ) | (
+                                ExecutionEnvironmentSelection::Restore,
+                                Compatible::RestoreSource(_)
+                            )
+                        ));
+                        return Ok(response.clone());
+                    }
+                    assert_eq!(intent, ExecutionEnvironmentSelection::ColdBoot);
+                    assert_eq!(
+                        request.reference,
+                        Some(exact_definition.reference().to_string())
+                    );
+                    if !exact_allowed {
+                        return Err(Status::failed_precondition("exact source unavailable"));
+                    }
+                    Ok(ResolveExecutionEnvironmentResponse {
+                        accepted_selection: 0,
+                        definition: Some((&exact_definition).into()),
+                        execution: Some((&exact_source).into()),
+                    })
+                },
+            )
+            .serve();
+        let service = SandboxManagerService::new_with_options(
+            crate::db::test_conn(),
+            SandboxManagerServiceOptions {
+                node_manager: node.connector(),
+                ..SandboxManagerServiceOptions::for_test()
+            },
+        );
+        let result = service.select_compatible_execution(&source, intent).await;
+        if let Some(expected) = expected {
+            assert_eq!(result.unwrap().runtime().version.as_str(), expected);
+        } else {
+            result.unwrap_err();
+        }
+        service.shutdown_token.cancel();
+        service.task_tracker.close();
+        tokio::time::timeout(
+            std::time::Duration::from_secs(30),
+            service.task_tracker.wait(),
+        )
+        .await
+        .unwrap();
+    }
+}
diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_core.rs b/crates/libs/sbx/mgr/server/src/service/grpc_core.rs
index 597863203..7530059c5 100644
--- a/crates/libs/sbx/mgr/server/src/service/grpc_core.rs
+++ b/crates/libs/sbx/mgr/server/src/service/grpc_core.rs
@@ -168,43 +168,16 @@ impl SandboxManagerService {
     async fn resolve_create_environment(
         &self,
         reference: pplx_space_common_core::types::EnvironmentReference,
-    ) -> Result<pplx_space_execution_common::environment::EnvironmentDefinition, Status> {
-        let response = tokio::time::timeout(std::time::Duration::from_secs(30), async {
+    ) -> Result<pplx_space_node_mgr_client::environment_selection::SelectedEnvironment, Status>
+    {
+        tokio::time::timeout(std::time::Duration::from_secs(30), async {
             let mut node = connect_node(&self.node_manager)
                 .await
                 .map_err(|error| Status::unavailable(error.to_string()))?;
-            node.resolve_execution_environment(
-                pplx_space_node_mgr_client::proto::ResolveExecutionEnvironmentRequest {
-                    compatible: None,
-                    reference: Some(reference.to_string()),
-                },
-            )
-            .await
+            super::environment_selection::resolve_exact(&mut node, reference).await
         })
         .await
-        .map_err(|_| Status::deadline_exceeded("environment admission timed out"))??
-        .into_inner();
-        let definition = pplx_space_execution_common::environment::EnvironmentDefinition::try_from(
-            response.definition.as_ref().ok_or_else(|| {
-                Status::failed_precondition("node did not return environment definition")
-            })?,
-        )
-        .map_err(|error| Status::failed_precondition(error.to_string()))?;
-        let actual = pplx_space_common_core::types::ExecutionMetadata::try_from(
-            response.execution.as_ref().ok_or_else(|| {
-                Status::failed_precondition("node did not acknowledge execution environment")
-            })?,
-        )
-        .map_err(|error| Status::failed_precondition(error.to_string()))?;
-        if definition.reference() != reference || actual.environment() != Some(reference) {
-            return Err(Status::failed_precondition(
-                "node returned a different or missing environment binding",
-            ));
-        }
-        definition
-            .bind_execution(actual)
-            .map_err(|error| Status::failed_precondition(error.to_string()))?;
-        Ok(definition)
+        .map_err(|_| Status::deadline_exceeded("environment admission timed out"))?
     }
 
     pub(super) async fn create(
@@ -241,10 +214,10 @@ impl SandboxManagerService {
                 super::validation::validate_environment_create_fields(req)?;
             }
             let verified = if let Some(reference) = environment {
-                let definition = self.resolve_create_environment(reference).await?;
-                let pplx_space_common_core::types::RuntimeBinding::ManagedRelease(release) = &definition.descriptor().spec().runtime;
+                let selected = self.resolve_create_environment(reference).await?;
+                let pplx_space_common_core::types::RuntimeBinding::ManagedRelease(release) = &selected.definition.descriptor().spec().runtime;
                 request.get_mut().runtime = release.runtime.kind.alias().to_owned();
-                Some(definition)
+                Some(selected)
             } else { None };
             let req = request.get_ref();
             let template = &req.template;
@@ -261,21 +234,14 @@ impl SandboxManagerService {
                 },
             )
             .await?;
-            if let Some(definition) = &verified {
-                resolved_template.execution = definition.bind_execution(resolved_template.execution)
+            if let Some(selected) = &verified {
+                selected.definition.check_cold_boot_source(&resolved_template.execution)
                     .map_err(|error| Status::failed_precondition(error.to_string()))?;
+                resolved_template.execution = selected.execution.clone();
             }
-            let non_default = super::validation::validate_environment_create(
+            super::validation::validate_environment_create(
                 req, environment, &resolved_template.execution,
             )?;
-            if non_default || environment.is_some() {
-                let reference = resolved_template.execution.environment().ok_or_else(|| Status::failed_precondition("explicit selection requires a bound template"))?;
-                if verified.is_none() {
-                    let definition = self.resolve_create_environment(reference).await?;
-                    resolved_template.execution = definition.bind_execution(resolved_template.execution)
-                        .map_err(|error| Status::failed_precondition(error.to_string()))?;
-                }
-            }
             if let Some(policy) = template_network_policy_from_proto(
                 req.template_network_policy.as_ref(),
                 req.template_network_policy_generation,
@@ -505,33 +471,25 @@ impl SandboxManagerService {
                         effective_mode,
                     )));
                 }
-                return self
-                    .create_with_id(CreateWithIdArgs {
-                        req,
-                        sandbox_owner: &sandbox_owner,
-                        effective_mode,
-                        id: requested_id.to_string(),
-                        template_dir: &resolved_template.template_dir,
-                        execution: resolved_template
-                            .execution
-                            .environment()
-                            .map(|_| &resolved_template.execution),
-                        snapshot_dir_resolved: resolved_template.snapshot_dir.as_deref(),
-                        snapshot_vcpu: resolved_template.snapshot_vcpu,
-                        content_hash_resolved: resolved_template.content_hash.as_deref(),
-                        auto_lifecycle: &auto_lifecycle,
-                        auto_lifecycle_authority,
-                        default_network_policy_authority: resolved_template
-                            .default_network_policy_authority,
-                    })
-                    .await;
+            }
+
+            // Reused IDs return above before selection. Cluster placement and
+            // pool identity still require their existing exact environment.
+            if environment.is_none()
+                && resolved_template.execution.environment().is_some()
+                && super::validation::allows_node_local_environment_selection(req)
+            {
+                resolved_template.execution = self.select_compatible_execution(
+                    &resolved_template.execution,
+                    pplx_space_node_mgr_client::proto::ExecutionEnvironmentSelection::ColdBoot,
+                ).await?;
             }
 
             self.create_with_id(CreateWithIdArgs {
                 req,
                 sandbox_owner: &sandbox_owner,
                 effective_mode,
-                id: Uuid::now_v7().to_string(),
+                id: requested_id.unwrap_or_else(Uuid::now_v7).to_string(),
                 template_dir: &resolved_template.template_dir,
                 execution: resolved_template
                     .execution
diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_core/create_after_lookup_tests.rs b/crates/libs/sbx/mgr/server/src/service/grpc_core/create_after_lookup_tests.rs
index a2437274c..198a99913 100644
--- a/crates/libs/sbx/mgr/server/src/service/grpc_core/create_after_lookup_tests.rs
+++ b/crates/libs/sbx/mgr/server/src/service/grpc_core/create_after_lookup_tests.rs
@@ -1,11 +1,8 @@
 //! Create's paths after template lookup, against a scripted template resolver
 //! and an in-process node-manager.
 //!
-//! A create that inserts a new sandbox row goes on to create the sandbox's
-//! cache root under the fixed host cache directory. Every service here
-//! refuses new sandbox rows with a SQLite trigger, so a create that reaches
-//! the insert stops there. The refusal carries the row create tried to
-//! write, which is what the cases that reach `create_with_id` assert on.
+//! Most cases fence the sandbox insert and inspect the attempted row. Successful
+//! logical creates instead use a test-owned cache root and assert durable state.
 
 use super::*;
 use crate::node_manager_transport::fake::{FakeNodeManager, ServedNodeManager};
@@ -803,6 +800,12 @@ async fn selected_create_persists_the_binding_used_by_first_start() {
         let execution = definition.execution_requirements().unwrap();
         let mut wire = pplx_space_execution_common::proto::ExecutionMetadata::from(&execution);
         wire.environment = None;
+        // The disk may have been built by an older consumer of the same
+        // kernel. The sandbox must keep the newly selected exact consumer.
+        if environment.kernel().version.as_str() == "6.1.155" {
+            wire.runtime.as_mut().unwrap().version = "1.15.1".into();
+            wire.state_format.as_mut().unwrap().version = "9.0.0".into();
+        }
         let legacy = pplx_space_common_core::types::ExecutionMetadata::try_from(&wire).unwrap();
         let build = TemplateBuild::new();
         let mut metadata: serde_json::Value =
@@ -890,7 +893,8 @@ async fn selected_create_persists_the_binding_used_by_first_start() {
         std::fs::write(build.dir.path().join("metadata.json"), &metadata_bytes).unwrap();
 
         // A retained rootfs and its matching legacy evidence also inherit the
-        // saved binding. No annotation may change runtime or kernel bytes.
+        // saved binding. Disk producer annotations cannot select a different
+        // consumer; conflicting guest or kernel inputs still fail.
         let sandbox_dir = cache.path().join(&created.sandbox_id);
         std::fs::create_dir_all(sandbox_dir.join("instance")).unwrap();
         std::fs::write(sandbox_dir.join("instance/rootfs.ext4"), b"test rootfs").unwrap();
@@ -899,6 +903,7 @@ async fn selected_create_persists_the_binding_used_by_first_start() {
             Matching,
             Alias,
             Runtime,
+            RuntimeKind,
             Kernel,
             Reference,
             Missing,
@@ -907,6 +912,7 @@ async fn selected_create_persists_the_binding_used_by_first_start() {
             Evidence::Matching,
             Evidence::Alias,
             Evidence::Runtime,
+            Evidence::RuntimeKind,
             Evidence::Kernel,
             Evidence::Reference,
             Evidence::Missing,
@@ -916,6 +922,10 @@ async fn selected_create_persists_the_binding_used_by_first_start() {
                 Evidence::Matching => {}
                 Evidence::Alias => wire.components[0].version = Some("legacy-kernel-label".into()),
                 Evidence::Runtime => wire.runtime.as_mut().unwrap().version = "1.16.0".into(),
+                Evidence::RuntimeKind => {
+                    wire.runtime.as_mut().unwrap().kind =
+                        pplx_space_execution_common::proto::RuntimeKind::CloudHypervisor as i32
+                }
                 Evidence::Kernel => {
                     wire.components[0].digest = Some(format!("sha256:{}", "00".repeat(32)))
                 }
@@ -945,7 +955,10 @@ async fn selected_create_persists_the_binding_used_by_first_start() {
                 cache.path(),
             )
             .await;
-            if matches!(evidence, Evidence::Matching | Evidence::Alias) {
+            if matches!(
+                evidence,
+                Evidence::Matching | Evidence::Alias | Evidence::Runtime | Evidence::Reference
+            ) {
                 let source = source.unwrap();
                 assert!(source.existing_artifact);
                 assert_eq!(source.execution.environment(), Some(definition.reference()));
@@ -969,3 +982,101 @@ async fn selected_create_persists_the_binding_used_by_first_start() {
             .unwrap();
     }
 }
+
+/// A revoked template source must reach compatible selection rather than first
+/// requiring exact permission for the revoked producer.
+#[tokio::test]
+async fn unpinned_create_selects_before_admitting_a_non_default_source() {
+    use pplx_space_common_core::types::{Arch, EnvironmentDescriptor, RuntimeBinding};
+    use pplx_space_execution_common::environment::EnvironmentDefinition;
+    use pplx_space_node_mgr_client::proto::{
+        ExecutionEnvironmentSelection, ResolveExecutionEnvironmentRequest,
+        ResolveExecutionEnvironmentResponse, resolve_execution_environment_request::Compatible,
+    };
+    if !enter_isolated_test_process() {
+        return;
+    }
+    let catalog =
+        pplx_space_node_runtime_core::catalog::CompiledEnvironment::for_architecture(Arch::X86_64)
+            .unwrap();
+    let definition = |name| {
+        let mut spec = catalog
+            .iter()
+            .find(|entry| entry.name() == name)
+            .unwrap()
+            .definition()
+            .descriptor()
+            .spec()
+            .clone();
+        let arch = std::env::consts::ARCH.parse().unwrap();
+        spec.host.architecture = arch;
+        spec.guest.architecture = arch;
+        let RuntimeBinding::ManagedRelease(runtime) = &mut spec.runtime;
+        runtime.target.architecture = arch;
+        EnvironmentDefinition::new(EnvironmentDescriptor::try_from(spec).unwrap())
+    };
+    let source = definition("fc-1.17.0-linux-6.1.155")
+        .execution_requirements()
+        .unwrap();
+    let target = definition("e0");
+    let execution = target.execution_requirements().unwrap();
+    let build = TemplateBuild::new();
+    let mut metadata: serde_json::Value =
+        serde_json::from_slice(&std::fs::read(build.dir.path().join("metadata.json")).unwrap())
+            .unwrap();
+    metadata["schema_version"] = 3.into();
+    metadata["execution"] = serde_json::to_value(&source).unwrap();
+    std::fs::write(
+        build.dir.path().join("metadata.json"),
+        serde_json::to_vec(&metadata).unwrap(),
+    )
+    .unwrap();
+    let expected_consumer = execution.clone();
+    let node = FakeNodeManager::default().unary(
+        "ResolveExecutionEnvironment",
+        move |request: ResolveExecutionEnvironmentRequest| {
+            let Some(Compatible::ColdBootSource(observed)) = request.compatible else {
+                return Err(Status::failed_precondition("source environment revoked"));
+            };
+            assert_eq!(
+                pplx_space_common_core::types::ExecutionMetadata::try_from(&observed).unwrap(),
+                source
+            );
+            Ok(ResolveExecutionEnvironmentResponse {
+                accepted_selection: ExecutionEnvironmentSelection::ColdBoot as i32,
+                definition: Some((&target).into()),
+                execution: Some((&execution).into()),
+            })
+        },
+    );
+    let database = tempfile::tempdir().unwrap();
+    let db_path = database.path().join("sandbox.sqlite");
+    crate::db::create_test_schema_on_disk(&db_path);
+    let mut fixture = Fixture::new(
+        crate::db::Connection::open_at(&db_path).unwrap(),
+        build.resolver(BUILD),
+        node,
+    );
+    fixture
+        .service
+        .db_conn()
+        .unwrap()
+        .execute_batch("DROP TRIGGER create_after_lookup_insert_fence")
+        .unwrap();
+    let cache = tempfile::tempdir().unwrap();
+    fixture.service.create_cache_dir = Some(cache.path().to_owned());
+    let mut request = create_request();
+    request.boot_mode = BootMode::Cold as i32;
+    let created = fixture.create(request).await.unwrap();
+    let reopened = crate::db::Connection::open_at(&db_path).unwrap();
+    let row = crate::db::get_sandbox(&reopened, &created.sandbox_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(row.execution_metadata, Some(expected_consumer));
+    assert_eq!(row.state, SandboxState::Created);
+    fixture.service.shutdown_token.cancel();
+    fixture.service.task_tracker.close();
+    tokio::time::timeout(CREATE_WAIT, fixture.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 5eb679c71..6be7245b6 100644
--- a/crates/libs/sbx/mgr/server/src/service/mod.rs
+++ b/crates/libs/sbx/mgr/server/src/service/mod.rs
@@ -1,3 +1,4 @@
+mod environment_selection;
 use pplx_space_common_core::types::NodeId;
 
 use std::collections::{BTreeMap, BTreeSet};
diff --git a/crates/libs/sbx/mgr/server/src/service/validation.rs b/crates/libs/sbx/mgr/server/src/service/validation.rs
index a027f9da9..7118c50d6 100644
--- a/crates/libs/sbx/mgr/server/src/service/validation.rs
+++ b/crates/libs/sbx/mgr/server/src/service/validation.rs
@@ -309,12 +309,7 @@ pub(super) fn validate_environment_create(
 pub(super) fn validate_environment_create_fields(
     request: &crate::proto::CreateRequest,
 ) -> Result<(), Status> {
-    if request.sandbox_id.is_some()
-        || request.route_identity.is_some()
-        || request.pool_template.is_some()
-        || request.snapshots_enabled
-        || request.boot_mode != crate::proto::BootMode::Cold as i32
-    {
+    if !allows_node_local_environment_selection(request) {
         return Err(Status::failed_precondition(
             "selected environments require node-local cold creation without pools, snapshots, or cluster-owned identity",
         ));
@@ -322,6 +317,18 @@ pub(super) fn validate_environment_create_fields(
     Ok(())
 }
 
+/// Node-local cold creation can select before its first persisted execution.
+/// Cluster-directed work and pools retain their placement-bound identities.
+pub(super) fn allows_node_local_environment_selection(
+    request: &crate::proto::CreateRequest,
+) -> bool {
+    request.sandbox_id.is_none()
+        && request.route_identity.is_none()
+        && request.pool_template.is_none()
+        && !request.snapshots_enabled
+        && request.boot_mode == crate::proto::BootMode::Cold as i32
+}
+
 /// Decode the additive input before normal admission. Both legacy inputs must
 /// be absent so an old server cannot ignore selection and execute default work.
 pub(super) fn normalize_environment_input(
diff --git a/docs/execution-environments.md b/docs/execution-environments.md
index 0e54e3f93..907699b07 100644
--- a/docs/execution-environments.md
+++ b/docs/execution-environments.md
@@ -598,14 +598,24 @@ in a versioned job payload. A downgraded worker cannot decode that payload as an
 ordinary build and silently substitute E0; such work must be inspected/resubmitted
 after upgrading, not treated as successful.
 
-An explicit selection must match the template's recorded environment. An unbound
-legacy template may acquire a binding for this creation only after its recorded
-runtime, guest, state format and kernel digest match the node-verified definition;
-its stored template metadata is not rewritten. Create commits the verified execution
-binding with the sandbox and cleanup intent before acknowledging success. First
-Start, service restarts, and later starts preserve it; conflicting or missing
-artifact evidence fails closed. Without a selection, a bound template retains
-that environment, including its exact runtime on new cold boot and stop/start.
+A selected cold-boot consumer must match the template's guest, runtime family
+and kernel digest. Its runtime version may differ from the template producer.
+The template's stored metadata is never relabelled. Create saves the selected
+execution with the sandbox and cleanup intent before acknowledging success.
+First Start, service restarts and later cold starts retain that exact consumer.
+Disk metadata records its producer; it cannot override the saved selection, and
+conflicting guest or kernel inputs fail. Snapshot producer evidence stays in the
+snapshot manifest, independently of the sandbox's current consumer.
+
+Unpinned scratch builds use the configured default, refusing when it is unusable.
+Derived builds and new node-local cold sandboxes prefer the bound source when
+allowed and usable. Only a revoked or unavailable source permits fallback to the
+newest allowed qualified consumer with the same kernel. Adding an allowed binding
+therefore does not upgrade a usable source or contaminate an exact benchmark arm.
+Explicit requests never fall back. Existing sandbox stop/start retains its saved
+consumer. After a failed restore that saved consumer may be the newly selected
+loader; its kernel identity is unchanged and cold admission rechecks policy.
+Cluster-directed work and pools still require their placement-bound identities.
 
 Retained instance directories grant sandbox-manager traversal without group write.
 Normal Start repairs old traversal permissions through node-manager before choosing
@@ -643,15 +653,19 @@ error before restarting. It repairs permissions on a valid directory, not a
 regular-file instance, symlink loop, or storage I/O fault; those conditions need
 filesystem repair without discarding the retained disk or rewriting its binding.
 
-Unbound derived builds keep the legacy request path regardless of runtime version.
-Ordinary builds inheriting the immutable E0 binding also retain that compatible
-path; explicit E0 requests and every inherited non-default binding require exact
-protocol acknowledgement. The distinction uses complete environment identity,
-never the runtime version alone. No kernel-only or runtime-only substitution is
-allowed. Local selected creation is cold-only, without pool adoption, snapshots,
-or cluster-owned identities. This creation restriction also covers explicit E0
-requests; later snapshot and suspend operations independently check memory-state
-qualification. Unqualified memory-state operations are refused.
+Unbound derived builds and custom kernel recipes retain their legacy contracts
+while the default-enabled `legacy` permission remains present.
+Every resolved build, including E0, sends its exact selected environment to Create
+so inventory changes cannot cause reselection. Local selected creation remains
+cold-only, without pool adoption, snapshots or cluster-owned identities; later
+snapshot and suspend operations independently check memory-state qualification.
+
+During mixed versions, an old resolver's missing selection acknowledgement keeps
+the pre-existing exact source contract. Scratch builds use ordinary default
+resolution, including during rollout. Other errors never cause fallback. A new
+non-default binding requires all its readers on the target node to be upgraded;
+older sandbox-managers reject cross-binding Create
+before execution. Roll out node-manager, then sandbox-manager, then builders.
 Managed non-default bindings are omitted from the legacy cluster placement
 projection. Historical external runtime and kernel installations remain available
 to legacy execution and placement. Fleet placement for complete environments,