node-manager: enforce and report explicit execution policy

crates/apps/node/mgr/cli/src/main.rs

@@ -314,6 +314,7 @@ async fn run(cli: Cli) -> Result<(), String> {
             let response = client
                 .resolve_execution_environment(
                     pplx_space_node_mgr_client::proto::ResolveExecutionEnvironmentRequest {
+                        compatible: None,
                         reference: reference.map(|value| value.to_string()),
                     },
                 )

crates/libs/api/src/routes/admin/node_store.rs

@@ -70,6 +70,9 @@ pub(super) struct AdminNodeRow {
     pub(super) metadata: Option<sqlx::types::Json<serde_json::Value>>,
     pub(super) last_heartbeat: Option<OffsetDateTime>,
     pub(super) status_observed_at: Option<OffsetDateTime>,
+    pub(super) execution_policy:
+        Option<sqlx::types::Json<pplx_space_common_core::types::ExecutionPolicyObservation>>,
+    pub(super) execution_policy_received_at: Option<OffsetDateTime>,
     pub(super) execution_inventory_status: Option<String>,
     pub(super) execution_inventory_received_at: Option<OffsetDateTime>,
     pub(super) status_health_healthy: Option<bool>,
@@ -285,6 +288,8 @@ const ADMIN_NODE_SELECT: &str = "SELECT n.id,
        nd.metadata,
        n.last_heartbeat,
        nss.observed_at AS status_observed_at,
+       nss.execution_policy,
+       nss.execution_policy_received_at,
        nss.execution_inventory_status,
        nss.execution_inventory_received_at,
        nss.health_healthy AS status_health_healthy,

crates/libs/api/src/routes/admin/nodes.rs

@@ -1,4 +1,4 @@
 use pplx_space_common_core::types::NodeId;
 
 use axum::{
     Extension, Json,
@@ -429,6 +429,8 @@
             .as_ref()
             .zip(row.execution_inventory_received_at)
             .map(|(status, received_at)| AdminExecutionInventoryObservation {
+                policy: row.execution_policy.as_ref().map(|value| value.0.clone()),
+                policy_received_at: row.execution_policy_received_at.map(format_time),
                 status: status.clone(),
                 received_at: format_time(received_at),
             }),

crates/libs/api/src/routes/admin/nodes/tests.rs

@@ -43,6 +43,8 @@ fn test_admin_node_row() -> AdminNodeRow {
         metadata: None,
         last_heartbeat: None,
         status_observed_at: None,
+        execution_policy: None,
+        execution_policy_received_at: None,
         execution_inventory_status: None,
         execution_inventory_received_at: None,
         status_health_healthy: None,

crates/libs/build/mgr/src/execution_binding.rs

@@ -112,6 +112,7 @@ impl BuildExecutionBinding {
             let mut node = pplx_space_node_mgr_client::connect().await?;
             Ok::<_, BuildExecutionBindingError>(
                 node.resolve_execution_environment(ResolveExecutionEnvironmentRequest {
+                    compatible: None,
                     reference: reference.map(|value| value.to_string()),
                 })
                 .await?

crates/libs/build/mgr/src/execution_binding/tests.rs

@@ -29,6 +29,7 @@ fn fixture() -> ResolveExecutionEnvironmentResponse {
         )
         .unwrap();
     ResolveExecutionEnvironmentResponse {
+        accepted_selection: 0,
         definition: Some((&definition).into()),
         execution: Some((&execution).into()),
     }
@@ -257,6 +258,7 @@ async fn derived_vm_requests_keep_unbound_releases_on_the_legacy_protocol() {
     let other = BuildExecutionBinding::from_response(
         None,
         ResolveExecutionEnvironmentResponse {
+            accepted_selection: 0,
             definition: Some((&definition).into()),
             execution: Some((&requirements).into()),
         },

crates/libs/common/core/src/types/admin.rs

@@ -353,6 +353,12 @@ pub struct AdminNodeStatusSnapshot {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct AdminExecutionInventoryObservation {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub policy: Option<super::ExecutionPolicyObservation>,
+    /// Independent receipt: older cluster writers may refresh inventory without
+    /// understanding policy. Their publication must not renew policy evidence.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub policy_received_at: Option<String>,
     /// Diagnostic label: complete, unavailable, or legacy. Preserve future
     /// labels for display during mixed-version rollout; never infer eligibility.
     pub status: String,

crates/libs/common/core/src/types/environment.rs

@@ -213,3 +213,14 @@ impl From<EnvironmentReference> for String {
         value.to_string()
     }
 }
+
+/// Observed eligibility is not proof that the artifacts are installed or usable.
+#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ExecutionPolicyObservation {
+    pub schema_version: u32,
+    pub allowed_environment_references: Vec<EnvironmentReference>,
+    pub default_environment_reference: Option<EnvironmentReference>,
+    #[cfg_attr(feature = "serde", serde(default))]
+    pub legacy_unbound_allowed: bool,
+}

crates/libs/execution/common/src/environment/binding.rs

@@ -53,6 +53,22 @@ impl EnvironmentDefinition {
         .with_environment(self.reference()))
     }
 
+    /// Check a cold-boot consumer against the source image contract. Runtime
+    /// version, state format and environment reference describe the producer;
+    /// only explicit compatible selection may replace those with a consumer.
+    /// This check is not installation evidence or execution authorization.
+    pub fn check_cold_boot_source(
+        &self,
+        source: &ExecutionMetadata,
+    ) -> Result<(), EnvironmentBindingError> {
+        let spec = self.descriptor().spec();
+        let RuntimeBinding::ManagedRelease(release) = &spec.runtime;
+        if source.runtime().kind != release.runtime.kind || source.guest() != spec.guest {
+            return Err(EnvironmentBindingError::RuntimeOrGuest);
+        }
+        self.check_boot_components(source)
+    }
+
     /// Bind observations only after checking their complete supported shape.
     /// Kernel bytes take precedence over historical directory-name aliases,
     /// matching the existing runtime component resolver's identity semantics.
@@ -87,6 +103,14 @@ impl EnvironmentDefinition {
         if execution.state_format() != Some(state_format) {
             return Err(EnvironmentBindingError::StateFormat);
         }
+        self.check_boot_components(execution)
+    }
+
+    fn check_boot_components(
+        &self,
+        execution: &ExecutionMetadata,
+    ) -> Result<(), EnvironmentBindingError> {
+        let spec = self.descriptor().spec();
         let BootProfile::LinuxDirectV1 { kernel } = &spec.boot;
         let [observed] = execution.components() else {
             return Err(EnvironmentBindingError::BootComponents);

crates/libs/execution/common/src/lib.rs

@@ -8,6 +8,7 @@ use pplx_space_common_core::types::{
 };
 
 pub mod environment;
+pub mod policy;
 
 pub mod proto {
     tonic::include_proto!("space.execution.v1");

crates/libs/execution/common/src/policy.rs

@@ -0,0 +1,63 @@
+//! Validate portable policy observations without assigning approval authority.
+use pplx_space_common_core::types::{DigestParseError, ExecutionPolicyObservation};
+
+#[derive(Debug, thiserror::Error)]
+pub enum PolicyObservationError {
+    #[error("unsupported execution policy schema {0}")]
+    Schema(u32),
+    #[error("invalid execution policy reference: {0}")]
+    Reference(#[from] DigestParseError),
+    #[error("execution policy has duplicate references or a default outside its allowlist")]
+    Shape,
+}
+
+impl TryFrom<&crate::proto::ExecutionPolicyObservation> for ExecutionPolicyObservation {
+    type Error = PolicyObservationError;
+    fn try_from(value: &crate::proto::ExecutionPolicyObservation) -> Result<Self, Self::Error> {
+        if value.schema_version != 1 {
+            return Err(PolicyObservationError::Schema(value.schema_version));
+        }
+        let allowed = value
+            .allowed_environment_references
+            .iter()
+            .map(|reference| reference.parse())
+            .collect::<Result<std::collections::BTreeSet<_>, _>>()?;
+        let default = value
+            .default_environment_reference
+            .as_deref()
+            .map(str::parse)
+            .transpose()?;
+        if allowed.len() != value.allowed_environment_references.len()
+            || default.is_some_and(|reference| !allowed.contains(&reference))
+            || (default.is_none() && !allowed.is_empty())
+        {
+            return Err(PolicyObservationError::Shape);
+        }
+        Ok(Self {
+            schema_version: 1,
+            legacy_unbound_allowed: value.legacy_unbound_allowed,
+            allowed_environment_references: allowed.into_iter().collect(),
+            default_environment_reference: default,
+        })
+    }
+}
+
+impl From<&ExecutionPolicyObservation> for crate::proto::ExecutionPolicyObservation {
+    fn from(value: &ExecutionPolicyObservation) -> Self {
+        Self {
+            schema_version: value.schema_version,
+            legacy_unbound_allowed: value.legacy_unbound_allowed,
+            allowed_environment_references: value
+                .allowed_environment_references
+                .iter()
+                .map(ToString::to_string)
+                .collect(),
+            default_environment_reference: value
+                .default_environment_reference
+                .map(|reference| reference.to_string()),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests;

crates/libs/execution/common/src/policy/tests.rs

@@ -0,0 +1,62 @@
+use super::*;
+use prost::Message;
+
+#[test]
+fn policy_observation_validates_shape_and_old_inventory_readers_omit_it() {
+    let reference = format!("sha256:{}", "11".repeat(32));
+    let wire = crate::proto::ExecutionPolicyObservation {
+        schema_version: 1,
+        legacy_unbound_allowed: true,
+        allowed_environment_references: vec![reference.clone()],
+        default_environment_reference: Some(reference.clone()),
+    };
+    let legacy_only = crate::proto::ExecutionPolicyObservation {
+        schema_version: 1,
+        legacy_unbound_allowed: true,
+        allowed_environment_references: vec![],
+        default_environment_reference: None,
+    };
+    let legacy_policy = ExecutionPolicyObservation::try_from(&legacy_only).unwrap();
+    assert!(legacy_policy.legacy_unbound_allowed);
+    assert_eq!(
+        crate::proto::ExecutionPolicyObservation::from(&legacy_policy),
+        legacy_only
+    );
+    let value = ExecutionPolicyObservation::try_from(&wire).unwrap();
+    assert_eq!(crate::proto::ExecutionPolicyObservation::from(&value), wire);
+    for bad in [
+        crate::proto::ExecutionPolicyObservation {
+            schema_version: 2,
+            ..wire.clone()
+        },
+        crate::proto::ExecutionPolicyObservation {
+            allowed_environment_references: vec![reference.clone(), reference],
+            ..wire.clone()
+        },
+        crate::proto::ExecutionPolicyObservation {
+            default_environment_reference: None,
+            ..wire.clone()
+        },
+        crate::proto::ExecutionPolicyObservation {
+            allowed_environment_references: vec![],
+            ..wire.clone()
+        },
+    ] {
+        ExecutionPolicyObservation::try_from(&bad).unwrap_err();
+    }
+    #[derive(Clone, PartialEq, Message)]
+    struct OldInventory {
+        #[prost(message, repeated, tag = "1")]
+        runtimes: Vec<crate::proto::RuntimeInstallation>,
+        #[prost(message, repeated, tag = "2")]
+        components: Vec<crate::proto::ArtifactComponentInstallation>,
+    }
+    let inventory = crate::proto::NodeExecutionInventory {
+        policy: Some(wire),
+        ..Default::default()
+    };
+    let old = OldInventory::decode(inventory.encode_to_vec().as_slice()).unwrap();
+    let forwarded =
+        crate::proto::NodeExecutionInventory::decode(old.encode_to_vec().as_slice()).unwrap();
+    assert!(forwarded.policy.is_none());
+}

crates/libs/internal/api/server/src/execution_observation.rs

@@ -104,10 +104,15 @@ pub(super) async fn prepare<'a>(
     Ok(Some(PreparedObservation { node_id, report }))
 }
 
+pub(super) struct PublishedExecutionObservation {
+    pub status: InventoryStatus,
+    pub policy: Option<pplx_space_common_core::types::ExecutionPolicyObservation>,
+}
+
 pub(super) async fn publish(
     tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
     prepared: Option<PreparedObservation<'_>>,
-) -> Result<Option<InventoryStatus>, Status> {
+) -> Result<Option<PublishedExecutionObservation>, Status> {
     let Some(PreparedObservation { node_id, report }) = prepared else {
         return Ok(None);
     };
@@ -119,7 +124,18 @@ pub(super) async fn publish(
     if let Some(inventory) = inventory {
         publish_inventory(tx, node_id, inventory).await?;
     }
-    Ok(Some(status))
+    let policy = inventory
+        .and_then(|inventory| inventory.policy.as_ref())
+        .and_then(|policy| {
+            match pplx_space_common_core::types::ExecutionPolicyObservation::try_from(policy) {
+                Ok(policy) => Some(policy),
+                Err(error) => {
+                    tracing::warn!(%error, %node_id, "invalid execution policy observation; withholding policy evidence");
+                    None
+                }
+            }
+        });
+    Ok(Some(PublishedExecutionObservation { status, policy }))
 }
 
 /// Reconcile a complete inventory without rewriting unchanged artifact rows.