fix(server): end image preparation on streamed configuration admission

Image preparation deadlines (#4038) leave preparation when a supervisor's
admission is recorded. ReportSandboxConfiguration did that, but admission
delivered over the supervisor session in push mode only stored the
admission, so every push-mode sandbox stayed Provisioning with
"Waiting for an authenticated supervisor configuration report" until its
deadline. Record the admission start, and any rejection, in the same CAS
that stores streamed admission.

Signed-off-by: Piotr Mlocek <pmlocek@nvidia.com>

Author: pimlock

crates/openshell-server/src/compute/mod.rs

@@ -4963,6 +4963,10 @@ impl ComputeRuntime {
             {
                 return Err("restarting sandbox requires a fresh supervisor instance".into());
             }
+            let provisioning = admission
+                .map(|admission| admitted_provisioning(&current, admission))
+                .transpose()?
+                .flatten();
             let expected_resource_version = sandbox_resource_version(&current);
             let result = self
                 .store
@@ -4979,6 +4983,9 @@ impl ComputeRuntime {
                                         == i32::from(openshell_core::proto::ConfigurationAdmissionState::Accepted),
                                 );
                             }
+                            if provisioning.is_some() {
+                                status.provisioning.clone_from(&provisioning);
+                            }
                             // An admission-only update leaves restart state to
                             // the instance that last reported runtime readiness.
                             if runtime_ready {
@@ -6738,6 +6745,31 @@ fn apply_driver_snapshot(
     apply_configuration_readiness(sandbox);
 }
 
+/// Admission evidence from a supervisor session ends image preparation, as a
+/// polled configuration report does. Returns the updated provisioning record,
+/// or `None` when the sandbox has none.
+fn admitted_provisioning(
+    sandbox: &Sandbox,
+    admission: &openshell_core::proto::SandboxConfigurationAdmission,
+) -> Result<Option<openshell_core::proto::SandboxProvisioning>, String> {
+    let Some(mut record) = sandbox
+        .status
+        .as_ref()
+        .and_then(|status| status.provisioning.clone())
+    else {
+        return Ok(None);
+    };
+    let now_ms = openshell_core::time::now_ms();
+    if !provisioning_deadline::allows_admission(&record, now_ms) {
+        return Err("provisioning deadline expired".to_string());
+    }
+    provisioning_deadline::record_admission_start(&mut record, now_ms)?;
+    if admission.state == i32::from(openshell_core::proto::ConfigurationAdmissionState::Rejected) {
+        provisioning_deadline::record_rejection(&mut record, now_ms)?;
+    }
+    Ok(Some(record))
+}
+
 /// Configuration readiness is independent of compute/container readiness.
 pub fn apply_configuration_readiness(sandbox: &mut Sandbox) {
     use openshell_core::proto::ConfigurationAdmissionState;

crates/openshell-server/src/grpc/policy.rs

@@ -8386,6 +8386,73 @@ mod tests {
         request
     }
 
+    #[tokio::test]
+    async fn streamed_admission_ends_preparation_before_runtime_readiness() {
+        use crate::compute::provisioning_deadline::new_preparation_record;
+        use openshell_core::proto::{
+            ConfigurationAdmissionState, SandboxConfigurationAdmission, SandboxPhase,
+        };
+        let state = test_server_state().await;
+        let sandbox_id = "sb-streamed-admission";
+        let mut sandbox = test_sandbox(
+            sandbox_id,
+            "streamed-admission",
+            openshell_policy::restrictive_default_policy(),
+            Vec::new(),
+        );
+        sandbox.set_phase(SandboxPhase::Provisioning.into());
+        sandbox.status.as_mut().unwrap().provisioning =
+            Some(new_preparation_record(current_time_ms(), 1800));
+        state.store.put_message(&sandbox).await.unwrap();
+        let instance_id = uuid::Uuid::new_v4().to_string();
+
+        // Push mode reports admission over the supervisor session instead of
+        // ReportSandboxConfiguration, then reports runtime readiness.
+        state
+            .compute
+            .supervisor_session_admission(
+                sandbox_id,
+                &instance_id,
+                &SandboxConfigurationAdmission {
+                    instance_id: instance_id.clone(),
+                    state: ConfigurationAdmissionState::Accepted.into(),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+        state
+            .compute
+            .supervisor_runtime_ready(sandbox_id, &instance_id)
+            .await
+            .unwrap();
+
+        let saved = state
+            .store
+            .get_message::<Sandbox>(sandbox_id)
+            .await
+            .unwrap()
+            .unwrap();
+        let status = saved.status.as_ref().unwrap();
+        assert!(
+            status
+                .provisioning
+                .as_ref()
+                .unwrap()
+                .admission_start_time
+                .is_some()
+        );
+        assert_eq!(saved.phase(), i32::from(SandboxPhase::Ready));
+        assert!(
+            status
+                .conditions
+                .iter()
+                .any(|condition| condition.r#type == "Ready" && condition.status == "True"),
+            "{:?}",
+            status.conditions
+        );
+    }
+
     #[tokio::test]
     async fn preparation_registration_starts_repair_once_after_a_cold_start() {
         use crate::compute::provisioning_deadline::new_preparation_record;