diff --git a/android/virtmgr/src/aidl.rs b/android/virtmgr/src/aidl.rs
index 5ad7ee1..57779bf 100644
--- a/android/virtmgr/src/aidl.rs
+++ b/android/virtmgr/src/aidl.rs
@@ -819,6 +819,8 @@
             .unwrap_or(Ok(UsbConfig { controller: false }))
             .or_binder_exception(ExceptionCode::BAD_PARCELABLE)?;
 
+        let detect_hangup = is_app_config && gdb_port.is_none();
+
         // Actually start the VM.
         let crosvm_config = CrosvmConfig {
             cid,
@@ -846,7 +848,7 @@
             ramdump,
             indirect_files,
             platform_version: parse_platform_version_req(&config.platformVersion)?,
-            detect_hangup: is_app_config,
+            detect_hangup,
             gdb_port,
             vfio_devices,
             dtbo,
diff --git a/build/debian/fai_config/scripts/AVF/10-systemd b/build/debian/fai_config/scripts/AVF/10-systemd
index a86d4c9..121acc5 100755
--- a/build/debian/fai_config/scripts/AVF/10-systemd
+++ b/build/debian/fai_config/scripts/AVF/10-systemd
@@ -3,9 +3,7 @@
 chmod +x $target/usr/local/bin/ttyd
 ln -s /etc/systemd/system/ttyd.service $target/etc/systemd/system/multi-user.target.wants/ttyd.service
 ln -s /etc/systemd/system/virtiofs.service $target/etc/systemd/system/multi-user.target.wants/virtiofs.service
-ln -s /etc/systemd/system/forwarder_guest_launcher.service $target/etc/systemd/system/multi-user.target.wants/forwarder_guest_launcher.service
 ln -s /etc/systemd/system/virtiofs_internal.service $target/etc/systemd/system/multi-user.target.wants/virtiofs_internal.service
 ln -s /etc/systemd/system/backup_mount.service $target/etc/systemd/system/multi-user.target.wants/backup_mount.service
-ln -s /etc/systemd/system/shutdown_runner.service $target/etc/systemd/system/multi-user.target.wants/shutdown_runner.service
 
 sed -i 's/#LLMNR=yes/LLMNR=no/' $target/etc/systemd/resolved.conf
diff --git a/guest/forwarder_guest_launcher/Cargo.toml b/guest/forwarder_guest_launcher/Cargo.toml
index 091d1cf..9c4d7e3 100644
--- a/guest/forwarder_guest_launcher/Cargo.toml
+++ b/guest/forwarder_guest_launcher/Cargo.toml
@@ -26,3 +26,5 @@
 maintainer = "ferrochrome-dev@google.com"
 copyright = "2024, The Android Open Source Project"
 depends = "$auto"
+maintainer-scripts = "debian/"
+systemd-units = { }
diff --git a/build/debian/fai_config/files/etc/systemd/system/forwarder_guest_launcher.service/AVF b/guest/forwarder_guest_launcher/debian/service
similarity index 98%
rename from build/debian/fai_config/files/etc/systemd/system/forwarder_guest_launcher.service/AVF
rename to guest/forwarder_guest_launcher/debian/service
index 61da1ce..6824c70 100644
--- a/build/debian/fai_config/files/etc/systemd/system/forwarder_guest_launcher.service/AVF
+++ b/guest/forwarder_guest_launcher/debian/service
@@ -3,6 +3,7 @@
 After=syslog.target
 After=network.target
 After=virtiofs_internal.service
+
 [Service]
 ExecStart=/usr/bin/bash -c '/usr/bin/forwarder_guest_launcher --grpc_port $(cat /mnt/internal/debian_service_port)'
 Type=simple
@@ -10,5 +11,6 @@
 RestartSec=1
 User=root
 Group=root
+
 [Install]
 WantedBy=multi-user.target
diff --git a/guest/microdroid_manager/aidl/android/system/virtualization/payload/IVmPayloadService.aidl b/guest/microdroid_manager/aidl/android/system/virtualization/payload/IVmPayloadService.aidl
index b7a539b..8d02d97 100644
--- a/guest/microdroid_manager/aidl/android/system/virtualization/payload/IVmPayloadService.aidl
+++ b/guest/microdroid_manager/aidl/android/system/virtualization/payload/IVmPayloadService.aidl
@@ -78,6 +78,20 @@
     byte[] getVmInstanceSecret(in byte[] identifier, int size);
 
     /**
+     * Write `data`, on behalf of the client, to Secretkeeper.
+     * This is confidential to the pVM and protected via appropriate DICE policy
+     * on the payload's DICE chain.
+     */
+    void writePayloadRpData(in byte[32] data);
+
+    /**
+     * Read payload's `data` written on behalf of the payload in Secretkeeper.
+     * The returned value can be null either due to no value written or because
+     * Android maliciously deleted the value - Secretkeeper deletion are not authenticated.
+     */
+    @nullable byte[32] readPayloadRpData();
+
+    /**
      * Gets the DICE attestation chain for the VM.
      *
      * The DICE chain must not be made available to all VMs as it contains privacy breaking
@@ -116,4 +130,13 @@
      *         certification chain.
      */
     AttestationResult requestAttestation(in byte[] challenge, in boolean testMode);
+
+    /**
+     * Checks whether the VM instance is new - i.e., if this is the first run of an instance.
+     * This is an indication of fresh new VM secrets. Payload can use this to setup the fresh
+     * instance if needed.
+     *
+     * @return true on the first boot of the instance & false on subsequent boot.
+     */
+    boolean isNewInstance();
 }
diff --git a/guest/microdroid_manager/src/main.rs b/guest/microdroid_manager/src/main.rs
index 451c3c8..57ad35d 100644
--- a/guest/microdroid_manager/src/main.rs
+++ b/guest/microdroid_manager/src/main.rs
@@ -250,7 +250,7 @@
 
     if is_strict_boot() {
         // Provisioning must happen on the first boot and never again.
-        if is_new_instance() {
+        if is_new_instance_legacy() {
             ensure!(
                 saved_data.is_none(),
                 MicrodroidError::PayloadInvalidConfig(
@@ -297,6 +297,17 @@
     Ok(instance_data)
 }
 
+// The VM instance run can be
+// 1. Either Newly created - which can happen if this is really a new VM instance (or a malicious
+//    Android has deleted relevant secrets)
+// 2. Or Re-run from an already seen VM instance.
+#[derive(PartialEq, Eq)]
+enum VmInstanceState {
+    Unknown,
+    NewlyCreated,
+    PreviouslySeen,
+}
+
 fn try_run_payload(
     service: &Strong<dyn IVirtualMachineService>,
     vm_payload_service_fd: OwnedFd,
@@ -326,8 +337,25 @@
     // To minimize the exposure to untrusted data, derive dice profile as soon as possible.
     info!("DICE derivation for payload");
     let dice_artifacts = dice_derivation(dice, &instance_data, &payload_metadata)?;
-    let vm_secret =
-        VmSecret::new(dice_artifacts, service).context("Failed to create VM secrets")?;
+    let mut state = VmInstanceState::Unknown;
+    let vm_secret = VmSecret::new(dice_artifacts, service, &mut state)
+        .context("Failed to create VM secrets")?;
+
+    let is_new_instance = match state {
+        VmInstanceState::NewlyCreated => true,
+        VmInstanceState::PreviouslySeen => false,
+        VmInstanceState::Unknown => {
+            // VmSecret instantiation was not able to determine the state. This should only happen
+            // for legacy secret mechanism (V1) - in which case fallback to legacy
+            // instance.img based determination of state.
+            ensure!(
+                !should_defer_rollback_protection(),
+                "VmInstanceState is Unknown whilst guest is expected to use V2 based secrets.
+                This should've never happened"
+            );
+            is_new_instance_legacy()
+        }
+    };
 
     if cfg!(dice_changes) {
         // Now that the DICE derivation is done, it's ok to allow payload code to run.
@@ -387,6 +415,7 @@
         service.clone(),
         vm_secret,
         vm_payload_service_fd,
+        is_new_instance,
     )?;
 
     // Set export_tombstones if enabled
@@ -488,7 +517,7 @@
     Path::new(AVF_STRICT_BOOT).exists()
 }
 
-fn is_new_instance() -> bool {
+fn is_new_instance_legacy() -> bool {
     Path::new(AVF_NEW_INSTANCE).exists()
 }
 
diff --git a/guest/microdroid_manager/src/vm_payload_service.rs b/guest/microdroid_manager/src/vm_payload_service.rs
index 7f4317b..fb57812 100644
--- a/guest/microdroid_manager/src/vm_payload_service.rs
+++ b/guest/microdroid_manager/src/vm_payload_service.rs
@@ -33,6 +33,7 @@
     allow_restricted_apis: bool,
     virtual_machine_service: Strong<dyn IVirtualMachineService>,
     secret: VmSecret,
+    is_new_instance: bool,
 }
 
 impl IVmPayloadService for VmPayloadService {
@@ -97,6 +98,29 @@
             certificateChain: cert_chain,
         })
     }
+
+    fn readPayloadRpData(&self) -> binder::Result<Option<[u8; 32]>> {
+        let data = self
+            .secret
+            .read_payload_data_rp()
+            .context("Failed to read payload's rollback protected data")
+            .with_log()
+            .or_service_specific_exception(-1)?;
+        Ok(data)
+    }
+
+    fn writePayloadRpData(&self, data: &[u8; 32]) -> binder::Result<()> {
+        self.secret
+            .write_payload_data_rp(data)
+            .context("Failed to write payload's rollback protected data")
+            .with_log()
+            .or_service_specific_exception(-1)?;
+        Ok(())
+    }
+
+    fn isNewInstance(&self) -> binder::Result<bool> {
+        Ok(self.is_new_instance)
+    }
 }
 
 impl Interface for VmPayloadService {}
@@ -107,8 +131,9 @@
         allow_restricted_apis: bool,
         vm_service: Strong<dyn IVirtualMachineService>,
         secret: VmSecret,
+        is_new_instance: bool,
     ) -> VmPayloadService {
-        Self { allow_restricted_apis, virtual_machine_service: vm_service, secret }
+        Self { allow_restricted_apis, virtual_machine_service: vm_service, secret, is_new_instance }
     }
 
     fn check_restricted_apis_allowed(&self) -> binder::Result<()> {
@@ -128,9 +153,10 @@
     vm_service: Strong<dyn IVirtualMachineService>,
     secret: VmSecret,
     vm_payload_service_fd: OwnedFd,
+    is_new_instance: bool,
 ) -> Result<()> {
     let vm_payload_binder = BnVmPayloadService::new_binder(
-        VmPayloadService::new(allow_restricted_apis, vm_service, secret),
+        VmPayloadService::new(allow_restricted_apis, vm_service, secret, is_new_instance),
         BinderFeatures::default(),
     );
 
diff --git a/guest/microdroid_manager/src/vm_secret.rs b/guest/microdroid_manager/src/vm_secret.rs
index 5cc90ff..56b3482 100644
--- a/guest/microdroid_manager/src/vm_secret.rs
+++ b/guest/microdroid_manager/src/vm_secret.rs
@@ -38,6 +38,7 @@
 use zeroize::Zeroizing;
 use std::sync::Mutex;
 use std::sync::Arc;
+use crate::VmInstanceState;
 
 const ENCRYPTEDSTORE_KEY_IDENTIFIER: &str = "encryptedstore_key";
 const AUTHORITY_HASH: i64 = -4670549;
@@ -71,13 +72,20 @@
     // with downgraded images will not have access to VM's secret.
     // V2 secrets require hardware support - Secretkeeper HAL, which (among other things)
     // is backed by tamper-evident storage, providing rollback protection to these secrets.
-    V2 { dice_artifacts: OwnedDiceArtifactsWithExplicitKey, skp_secret: ZVec },
+    V2 {
+        instance_id: [u8; ID_SIZE],
+        dice_artifacts: OwnedDiceArtifactsWithExplicitKey,
+        skp_secret: ZVec,
+        secretkeeper_session: SkVmSession,
+    },
     // V1 secrets are not protected against rollback of boot images.
     // They are reliable only if rollback of images was prevented by verified boot ie,
     // each stage (including pvmfw/Microdroid/Microdroid Manager) prevents downgrade of next
     // stage. These are now legacy secrets & used only when Secretkeeper HAL is not supported
     // by device.
-    V1 { dice_artifacts: OwnedDiceArtifacts },
+    V1 {
+        dice_artifacts: OwnedDiceArtifacts,
+    },
 }
 
 // For supporting V2 secrets, guest expects the public key to be present in the Linux device tree.
@@ -92,6 +100,7 @@
     pub fn new(
         dice_artifacts: OwnedDiceArtifacts,
         vm_service: &Strong<dyn IVirtualMachineService>,
+        state: &mut VmInstanceState,
     ) -> Result<Self> {
         ensure!(dice_artifacts.bcc().is_some(), "Dice chain missing");
         if !crate::should_defer_rollback_protection() {
@@ -100,24 +109,28 @@
 
         let explicit_dice = OwnedDiceArtifactsWithExplicitKey::from_owned_artifacts(dice_artifacts)
             .context("Failed to get Dice artifacts in explicit key format")?;
-        let session = SkVmSession::new(vm_service, &explicit_dice)?;
         let id = super::get_instance_id()?.ok_or(anyhow!("Missing instance_id"))?;
         let explicit_dice_chain = explicit_dice
             .explicit_key_dice_chain()
             .ok_or(anyhow!("Missing explicit dice chain, this is unusual"))?;
         let policy = sealing_policy(explicit_dice_chain)
             .map_err(|e| anyhow!("Failed to build a sealing_policy: {e}"))?;
+        let session = SkVmSession::new(vm_service, &explicit_dice, policy)?;
         let mut skp_secret = Zeroizing::new([0u8; SECRET_SIZE]);
-        if let Some(secret) = session.get_secret(id, Some(policy.clone()))? {
-            *skp_secret = secret
+        if let Some(secret) = session.get_secret(id)? {
+            *skp_secret = secret;
+            *state = VmInstanceState::PreviouslySeen;
         } else {
             log::warn!("No entry found in Secretkeeper for this VM instance, creating new secret.");
             *skp_secret = rand::random();
-            session.store_secret(id, skp_secret.clone(), policy)?;
+            session.store_secret(id, skp_secret.clone())?;
+            *state = VmInstanceState::NewlyCreated;
         }
         Ok(Self::V2 {
+            instance_id: id,
             dice_artifacts: explicit_dice,
             skp_secret: ZVec::try_from(skp_secret.to_vec())?,
+            secretkeeper_session: session,
         })
     }
 
@@ -130,7 +143,7 @@
 
     fn get_vm_secret(&self, salt: &[u8], identifier: &[u8], key: &mut [u8]) -> Result<()> {
         match self {
-            Self::V2 { dice_artifacts, skp_secret } => {
+            Self::V2 { dice_artifacts, skp_secret, .. } => {
                 let mut hasher = sha::Sha256::new();
                 hasher.update(dice_artifacts.cdi_seal());
                 hasher.update(skp_secret);
@@ -152,6 +165,23 @@
     pub fn derive_encryptedstore_key(&self, key: &mut [u8]) -> Result<()> {
         self.get_vm_secret(SALT_ENCRYPTED_STORE, ENCRYPTEDSTORE_KEY_IDENTIFIER.as_bytes(), key)
     }
+
+    pub fn read_payload_data_rp(&self) -> Result<Option<[u8; SECRET_SIZE]>> {
+        let Self::V2 { instance_id, secretkeeper_session, .. } = self else {
+            return Err(anyhow!("Rollback protected data is not available with V1 secrets"));
+        };
+        let payload_id = sha::sha512(instance_id);
+        secretkeeper_session.get_secret(payload_id)
+    }
+
+    pub fn write_payload_data_rp(&self, data: &[u8; SECRET_SIZE]) -> Result<()> {
+        let data = Zeroizing::new(*data);
+        let Self::V2 { instance_id, secretkeeper_session, .. } = self else {
+            return Err(anyhow!("Rollback protected data is not available with V1 secrets"));
+        };
+        let payload_id = sha::sha512(instance_id);
+        secretkeeper_session.store_secret(payload_id, data)
+    }
 }
 
 // Construct a sealing policy on the dice chain. VMs uses the following set of constraint for
@@ -227,31 +257,35 @@
 }
 
 // The secure session between VM & Secretkeeper
-struct SkVmSession(Arc<Mutex<SkSession>>);
+pub(crate) struct SkVmSession {
+    session: Arc<Mutex<SkSession>>,
+    sealing_policy: Vec<u8>,
+}
+
+// TODO(b/378911776): This get_secret/store_secret fails on expired session.
+// Introduce retry after refreshing the session
 impl SkVmSession {
     fn new(
         vm_service: &Strong<dyn IVirtualMachineService>,
         dice: &OwnedDiceArtifactsWithExplicitKey,
+        sealing_policy: Vec<u8>,
     ) -> Result<Self> {
         let secretkeeper_proxy = get_secretkeeper_service(vm_service)?;
-        let secure_session =
-            SkSession::new(secretkeeper_proxy, dice, Some(get_secretkeeper_identity()?))?;
-        let secure_session = Arc::new(Mutex::new(secure_session));
-        Ok(Self(secure_session))
+        let session = SkSession::new(secretkeeper_proxy, dice, Some(get_secretkeeper_identity()?))?;
+        let session = Arc::new(Mutex::new(session));
+        Ok(Self { session, sealing_policy })
     }
 
-    fn store_secret(
-        &self,
-        id: [u8; ID_SIZE],
-        secret: Zeroizing<[u8; SECRET_SIZE]>,
-        sealing_policy: Vec<u8>,
-    ) -> Result<()> {
-        let store_request =
-            StoreSecretRequest { id: Id(id), secret: Secret(*secret), sealing_policy };
+    fn store_secret(&self, id: [u8; ID_SIZE], secret: Zeroizing<[u8; SECRET_SIZE]>) -> Result<()> {
+        let store_request = StoreSecretRequest {
+            id: Id(id),
+            secret: Secret(*secret),
+            sealing_policy: self.sealing_policy.clone(),
+        };
         log::info!("Secretkeeper operation: {:?}", store_request);
 
         let store_request = store_request.serialize_to_packet().to_vec().map_err(anyhow_err)?;
-        let session = &mut *self.0.lock().unwrap();
+        let session = &mut *self.session.lock().unwrap();
         let store_response = session.secret_management_request(&store_request)?;
         let store_response = ResponsePacket::from_slice(&store_response).map_err(anyhow_err)?;
         let response_type = store_response.response_type().map_err(anyhow_err)?;
@@ -263,15 +297,14 @@
         Ok(())
     }
 
-    fn get_secret(
-        &self,
-        id: [u8; ID_SIZE],
-        updated_sealing_policy: Option<Vec<u8>>,
-    ) -> Result<Option<[u8; SECRET_SIZE]>> {
-        let get_request = GetSecretRequest { id: Id(id), updated_sealing_policy };
+    fn get_secret(&self, id: [u8; ID_SIZE]) -> Result<Option<[u8; SECRET_SIZE]>> {
+        let get_request = GetSecretRequest {
+            id: Id(id),
+            updated_sealing_policy: Some(self.sealing_policy.clone()),
+        };
         log::info!("Secretkeeper operation: {:?}", get_request);
         let get_request = get_request.serialize_to_packet().to_vec().map_err(anyhow_err)?;
-        let session = &mut *self.0.lock().unwrap();
+        let session = &mut *self.session.lock().unwrap();
         let get_response = session.secret_management_request(&get_request)?;
         let get_response = ResponsePacket::from_slice(&get_response).map_err(anyhow_err)?;
         let response_type = get_response.response_type().map_err(anyhow_err)?;
diff --git a/guest/shutdown_runner/Cargo.toml b/guest/shutdown_runner/Cargo.toml
index 564daf6..0b44baa 100644
--- a/guest/shutdown_runner/Cargo.toml
+++ b/guest/shutdown_runner/Cargo.toml
@@ -20,3 +20,5 @@
 maintainer = "ferrochrome-dev@google.com"
 copyright = "2024, The Android Open Source Project"
 depends = "$auto"
+maintainer-scripts = "debian/"
+systemd-units = { }
diff --git a/build/debian/fai_config/files/etc/systemd/system/shutdown_runner.service/AVF b/guest/shutdown_runner/debian/service
similarity index 98%
rename from build/debian/fai_config/files/etc/systemd/system/shutdown_runner.service/AVF
rename to guest/shutdown_runner/debian/service
index 8656c72..d19ff25 100644
--- a/build/debian/fai_config/files/etc/systemd/system/shutdown_runner.service/AVF
+++ b/guest/shutdown_runner/debian/service
@@ -2,10 +2,12 @@
 After=syslog.target
 After=network.target
 After=virtiofs_internal.service
+
 [Service]
 ExecStart=/usr/bin/bash -c '/usr/bin/shutdown_runner --grpc_port $(cat /mnt/internal/debian_service_port)'
 Type=simple
 User=root
 Group=root
+
 [Install]
 WantedBy=multi-user.target
diff --git a/libs/devicemapper/src/loopdevice.rs b/libs/devicemapper/src/loopdevice.rs
index 113a946..130c1c4 100644
--- a/libs/devicemapper/src/loopdevice.rs
+++ b/libs/devicemapper/src/loopdevice.rs
@@ -159,6 +159,7 @@
 #[cfg(test)]
 mod tests {
     use super::*;
+    use rdroidtest::rdroidtest;
     use std::fs;
     use std::path::Path;
 
@@ -178,7 +179,7 @@
         "0" == fs::read_to_string(ro).unwrap().trim()
     }
 
-    #[test]
+    #[rdroidtest]
     fn attach_loop_device_with_dio() {
         let a_dir = tempfile::TempDir::new().unwrap();
         let a_file = a_dir.path().join("test");
@@ -191,7 +192,7 @@
         assert!(is_direct_io(&dev));
     }
 
-    #[test]
+    #[rdroidtest]
     fn attach_loop_device_without_dio() {
         let a_dir = tempfile::TempDir::new().unwrap();
         let a_file = a_dir.path().join("test");
@@ -204,7 +205,7 @@
         assert!(!is_direct_io(&dev));
     }
 
-    #[test]
+    #[rdroidtest]
     fn attach_loop_device_with_dio_writable() {
         let a_dir = tempfile::TempDir::new().unwrap();
         let a_file = a_dir.path().join("test");
diff --git a/libs/libvm_payload/Android.bp b/libs/libvm_payload/Android.bp
index bb91737..1ebbe39 100644
--- a/libs/libvm_payload/Android.bp
+++ b/libs/libvm_payload/Android.bp
@@ -34,6 +34,7 @@
     bindgen_flags: [
         "--default-enum-style rust",
         "--allowlist-type=AVmAttestationStatus",
+        "--allowlist-type=AVmAccessRollbackProtectedSecretStatus",
     ],
     visibility: [":__subpackages__"],
 }
diff --git a/libs/libvm_payload/include/vm_payload.h b/libs/libvm_payload/include/vm_payload.h
index 5e15607..e4609fa 100644
--- a/libs/libvm_payload/include/vm_payload.h
+++ b/libs/libvm_payload/include/vm_payload.h
@@ -52,6 +52,22 @@
 } AVmAttestationStatus;
 
 /**
+ * Introduced in API 36.
+ * Status type used to indicate error while accessing RollbackProtectedSecret.
+ */
+typedef enum AVmAccessRollbackProtectedSecretStatus : int32_t {
+    /**
+     * Relevant Entry not found. This can happen either due to no value was ever written or because
+     * Android maliciously deleted the value (deletions may not be authenticated).
+     */
+    AVMACCESSROLLBACKPROTECTEDSECRETSTATUS_ENTRY_NOT_FOUND = -1,
+    /** Requested access size is not supported by the implementation */
+    AVMACCESSROLLBACKPROTECTEDSECRETSTATUS_BAD_SIZE = -2,
+    /** Access failed, this could be due to lacking support from Hardware */
+    AVMACCESSROLLBACKPROTECTEDSECRETSTATUS_ACCESS_FAILED = -3,
+} AVmAccessRollbackProtectedSecretStatus;
+
+/**
  * Notifies the host that the payload is ready.
  *
  * If the host app has set a `VirtualMachineCallback` for the VM, its
@@ -102,6 +118,11 @@
  * byte sequences and do not need to be kept secret; typically they are
  * hardcoded in the calling code.
  *
+ * The secret is linked to the instance & will be created for a new instance.
+ * Callers should check `AVmPayload_isNewInstance()` to meaningfully use the secret.
+ * For ex, decryption of any old data is meaningless with the returned secret of a new
+ * VM instance with fresh keys.
+ *
  * \param identifier identifier of the secret to return.
  * \param identifier_size size of the secret identifier.
  * \param secret pointer to size bytes where the secret is written.
@@ -259,5 +280,42 @@
 size_t AVmAttestationResult_getCertificateAt(const AVmAttestationResult* _Nonnull result,
                                              size_t index, void* _Nullable data, size_t size)
         __INTRODUCED_IN(__ANDROID_API_V__);
+/**
+ * Writes up to n bytes from buffer starting at `buf`, on behalf of the payload, to rollback
+ * detectable storage. The number of bytes written may be less than n if, for example, the
+ * underlying storage has size constraints. This stored data is confidential to the pVM and
+ * protected via appropriate DICE policy on the payload's DICE chain.
+ *
+ * \param buf A pointer to data to be written. This should have the size of at least n bytes.
+ * \param n The maximum number of bytes to be filled in `buf`.
+ *
+ *  \return On success, the number of bytes written is returned. On error, appropriate
+ * AVmAccessRollbackProtectedSecretStatus (negative number) is returned.
+ */
+
+int32_t AVmPayload_writeRollbackProtectedSecret(const void* _Nonnull buf, size_t n)
+        __INTRODUCED_IN(36);
+/**
+ * Read up to n bytes of payload's data in rollback detectable storage into `buf`.
+ *
+ * \param buf A pointer to buffer where the requested data is written. This should have the size of
+ * at least n bytes.
+ * \param n The maximum number of bytes to be read.
+ *
+ *  \return On success, the number of bytes that would have been written to `buf` if n was
+ * sufficiently large. On error, appropriate AVmAccessRollbackProtectedSecretStatus(a negative
+ * number) is returned.
+ */
+int32_t AVmPayload_readRollbackProtectedSecret(void* _Nullable buf, size_t n) __INTRODUCED_IN(36);
+;
+
+/**
+ * Checks whether the VM instance is new - i.e., if this is the first run of an instance.
+ * This is an indication of fresh new VM secrets. Payload can use this to setup the fresh
+ * instance if needed.
+ *
+ *  \return true if this is the first run of an instance, false otherwise.
+ */
+bool AVmPayload_isNewInstance() __INTRODUCED_IN(36);
 
 __END_DECLS
diff --git a/libs/libvm_payload/libvm_payload.map.txt b/libs/libvm_payload/libvm_payload.map.txt
index 3daad00..ca949d9 100644
--- a/libs/libvm_payload/libvm_payload.map.txt
+++ b/libs/libvm_payload/libvm_payload.map.txt
@@ -15,6 +15,9 @@
     AVmAttestationStatus_toString;       # systemapi introduced=VanillaIceCream
     AVmAttestationResult_getCertificateCount; # systemapi introduced=VanillaIceCream
     AVmAttestationResult_getCertificateAt; # systemapi introduced=VanillaIceCream
+    AVmPayload_writeRollbackProtectedSecret; # systemapi introduced=36
+    AVmPayload_readRollbackProtectedSecret; # systemapi introduced=36
+    AVmPayload_isNewInstance;                # systemapi introduced=36
   local:
     *;
 };
diff --git a/libs/libvm_payload/src/lib.rs b/libs/libvm_payload/src/lib.rs
index eb81752..cbadec2 100644
--- a/libs/libvm_payload/src/lib.rs
+++ b/libs/libvm_payload/src/lib.rs
@@ -16,14 +16,14 @@
 
 use android_system_virtualization_payload::aidl::android::system::virtualization::payload:: IVmPayloadService::{
     IVmPayloadService, ENCRYPTEDSTORE_MOUNTPOINT, VM_APK_CONTENTS_PATH,
-    VM_PAYLOAD_SERVICE_SOCKET_NAME, AttestationResult::AttestationResult,
+    VM_PAYLOAD_SERVICE_SOCKET_NAME, AttestationResult::AttestationResult
 };
 use anyhow::{bail, ensure, Context, Result};
 use binder::{
     unstable_api::{new_spibinder, AIBinder},
     Strong, ExceptionCode,
 };
-use log::{error, info, LevelFilter};
+use log::{error, info, LevelFilter, debug};
 use rpcbinder::{RpcServer, RpcSession};
 use openssl::{ec::EcKey, sha::sha256, ecdsa::EcdsaSig};
 use std::convert::Infallible;
@@ -38,9 +38,12 @@
     Mutex,
 };
 use vm_payload_status_bindgen::AVmAttestationStatus;
+use vm_payload_status_bindgen::AVmAccessRollbackProtectedSecretStatus::{AVMACCESSROLLBACKPROTECTEDSECRETSTATUS_ENTRY_NOT_FOUND, AVMACCESSROLLBACKPROTECTEDSECRETSTATUS_ACCESS_FAILED, AVMACCESSROLLBACKPROTECTEDSECRETSTATUS_BAD_SIZE};
+use std::cmp::min;
 
 /// Maximum size of an ECDSA signature for EC P-256 key is 72 bytes.
 const MAX_ECDSA_P256_SIGNATURE_SIZE: usize = 72;
+const RP_DATA_SIZE: usize = 32;
 
 static VM_APK_CONTENTS_PATH_C: LazyLock<CString> =
     LazyLock::new(|| CString::new(VM_APK_CONTENTS_PATH).expect("CString::new failed"));
@@ -566,3 +569,98 @@
         ptr::null()
     }
 }
+
+/// Writes up to n bytes from buffer starting at `buf`, on behalf of the payload, to rollback
+/// detectable storage and return the number of bytes written or appropriate (negative) status.
+/// For this implementation, the backing storage is Secretkeeper HAL, which allows storing & reading
+/// of 32 bytes secret!
+///
+/// # Safety
+///
+/// Behavior is undefined if any of the following conditions are violated:
+///
+/// * `buf` must be [valid] for reads of n bytes.
+///
+/// [valid]: ptr#safety
+#[no_mangle]
+pub unsafe extern "C" fn AVmPayload_writeRollbackProtectedSecret(buf: *const u8, n: usize) -> i32 {
+    initialize_logging();
+    if n < RP_DATA_SIZE {
+        error!(
+            "Requested writing {} bytes, while Secretkeeper supports only {} bytes",
+            n, RP_DATA_SIZE
+        );
+        return AVMACCESSROLLBACKPROTECTEDSECRETSTATUS_BAD_SIZE as i32;
+    }
+    // Safety: See the requirements on `buf` above and we just checked that n >= RP_DATA_SIZE.
+    let buf = unsafe { std::slice::from_raw_parts(buf, RP_DATA_SIZE) };
+    match try_writing_payload_rollback_protected_data(buf.try_into().unwrap()) {
+        Ok(()) => RP_DATA_SIZE as i32,
+        Err(e) => {
+            error!("Failed to write rollback protected data: {e:?}");
+            AVMACCESSROLLBACKPROTECTEDSECRETSTATUS_ACCESS_FAILED as i32
+        }
+    }
+}
+
+/// Read up to n bytes of payload's data in rollback detectable storage into `buf`.
+/// For this implementation, the backing storage is Secretkeeper HAL, which allows storing & reading
+/// of 32 bytes secret!
+///
+/// # Safety
+///
+/// Behavior is undefined if any of the following conditions are violated:
+///
+/// * `buf` must be [valid] for writes of n bytes.
+///
+/// [valid]: ptr#safety
+#[no_mangle]
+pub unsafe extern "C" fn AVmPayload_readRollbackProtectedSecret(buf: *mut u8, n: usize) -> i32 {
+    initialize_logging();
+    match try_read_rollback_protected_data() {
+        Err(e) => {
+            error!("Failed to read rollback protected data: {e:?}");
+            AVMACCESSROLLBACKPROTECTEDSECRETSTATUS_ACCESS_FAILED as i32
+        }
+        Ok(stored_data) => {
+            if let Some(stored_data) = stored_data {
+                // SAFETY: See the requirements on `buf` above; `stored_data` is known to have
+                // length `RP_DATA_SIZE`, and cannot overlap `data` because we just allocated
+                // it.
+                unsafe {
+                    ptr::copy_nonoverlapping(stored_data.as_ptr(), buf, min(n, RP_DATA_SIZE));
+                }
+                RP_DATA_SIZE as i32
+            } else {
+                debug!("No relevant entry found in Secretkeeper");
+                AVMACCESSROLLBACKPROTECTEDSECRETSTATUS_ENTRY_NOT_FOUND as i32
+            }
+        }
+    }
+}
+
+fn try_writing_payload_rollback_protected_data(data: &[u8; RP_DATA_SIZE]) -> Result<()> {
+    get_vm_payload_service()?
+        .writePayloadRpData(data)
+        .context("Failed to write payload rollback protected data")?;
+    Ok(())
+}
+
+fn try_read_rollback_protected_data() -> Result<Option<[u8; RP_DATA_SIZE]>> {
+    let rp = get_vm_payload_service()?
+        .readPayloadRpData()
+        .context("Failed to read rollback protected data")?;
+    Ok(rp)
+}
+
+/// Checks whether the VM instance is new - i.e., if this is the first run of an instance.
+///
+/// Panics on error (including unexpected server exit).
+#[no_mangle]
+pub extern "C" fn AVmPayload_isNewInstance() -> bool {
+    unwrap_or_abort(try_is_new_instance())
+}
+
+fn try_is_new_instance() -> Result<bool> {
+    get_vm_payload_service()?.isNewInstance().context("Cannot determine if the instance is new")
+}
diff --git a/libs/libvm_payload/wrapper/lib.rs b/libs/libvm_payload/wrapper/lib.rs
index b9ce6c8..bf274b0 100644
--- a/libs/libvm_payload/wrapper/lib.rs
+++ b/libs/libvm_payload/wrapper/lib.rs
@@ -31,7 +31,9 @@
 use std::ptr;
 use vm_payload_bindgen::{
     AIBinder, AVmPayload_getApkContentsPath, AVmPayload_getEncryptedStoragePath,
-    AVmPayload_getVmInstanceSecret, AVmPayload_notifyPayloadReady, AVmPayload_runVsockRpcServer,
+    AVmPayload_getVmInstanceSecret, AVmPayload_isNewInstance, AVmPayload_notifyPayloadReady,
+    AVmPayload_readRollbackProtectedSecret, AVmPayload_runVsockRpcServer,
+    AVmPayload_writeRollbackProtectedSecret,
 };
 
 /// The functions declared here are restricted to VMs created with a config file;
@@ -194,3 +196,23 @@
         )
     }
 }
+
+/// Read payload's `data` written on behalf of the payload in Secretkeeper.
+pub fn read_rollback_protected_secret(data: &mut [u8]) -> i32 {
+    // SAFETY: The function only reads from`[data]` within its bounds.
+    unsafe { AVmPayload_readRollbackProtectedSecret(data.as_ptr() as *mut c_void, data.len()) }
+}
+
+/// Write `data`, on behalf of the payload, to Secretkeeper.
+pub fn write_rollback_protected_secret(data: &[u8]) -> i32 {
+    // SAFETY: The function only writes to `[data]` within its bounds.
+    unsafe { AVmPayload_writeRollbackProtectedSecret(data.as_ptr() as *const c_void, data.len()) }
+}
+
+/// Checks whether the VM instance is new - i.e., if this is the first run of an instance.
+/// This is an indication of fresh new VM secrets. Payload can use this to setup the fresh
+/// instance if needed.
+pub fn is_new_instance_status() -> bool {
+    // SAFETY: The function returns bool, no arguments are needed.
+    unsafe { AVmPayload_isNewInstance() }
+}
diff --git a/tests/aidl/com/android/microdroid/testservice/ITestService.aidl b/tests/aidl/com/android/microdroid/testservice/ITestService.aidl
index 6a3bc1b..6a413d6 100644
--- a/tests/aidl/com/android/microdroid/testservice/ITestService.aidl
+++ b/tests/aidl/com/android/microdroid/testservice/ITestService.aidl
@@ -80,8 +80,28 @@
     String readLineFromConsole();
 
     /**
+     * Read payload's rollback protected data. The `AVmAccessRollbackProtectedSecretStatus` is
+     * wrapped as service_specific error in case of failure. This is _only_ used for testing.
+     */
+    byte[32] insecurelyReadPayloadRpData();
+
+    /**
+     * Request VM to write payload's rollback protected data. The
+     * `AVmAccessRollbackProtectedSecretStatus` is wrapped as service_specific error in case of
+     * failure. This is _only_ used for testing.
+     */
+    void insecurelyWritePayloadRpData(in byte[32] data);
+
+    /**
      * Request the service to exit, triggering the termination of the VM. This may cause any
      * requests in flight to fail.
      */
     oneway void quit();
+
+    /**
+     * Checks whether the VM instance is new - i.e., if this is the first run of an instance.
+     *
+     * @return true on the first boot of the instance & false on subsequent boot.
+     */
+    boolean isNewInstance();
 }
diff --git a/tests/early_vm_test/Android.bp b/tests/early_vm_test/Android.bp
new file mode 100644
index 0000000..dbb0c28
--- /dev/null
+++ b/tests/early_vm_test/Android.bp
@@ -0,0 +1,53 @@
+prebuilt_etc {
+    name: "avf_early_vm_test_kernel",
+    filename: "rialto.bin",
+    src: ":empty_file",
+    target: {
+        android_arm64: {
+            src: ":rialto_signed",
+        },
+    },
+    installable: false,
+    system_ext_specific: true,
+    visibility: ["//visibility:private"],
+}
+
+rust_binary {
+    name: "avf_early_vm_test_launcher",
+    crate_name: "avf_early_vm_test_launcher",
+    srcs: ["src/main.rs"],
+    rustlibs: [
+        "android.system.virtualizationservice-rust",
+        "libanyhow",
+        "libclap",
+        "libhypervisor_props",
+        "liblog_rust",
+        "libservice_vm_comm",
+        "libservice_vm_manager",
+        "libvmclient",
+    ],
+    cfgs: select(release_flag("RELEASE_AVF_ENABLE_EARLY_VM"), {
+        true: ["early_vm_enabled"],
+        default: [],
+    }),
+    prefer_rlib: true,
+    system_ext_specific: true,
+    compile_multilib: "first",
+    installable: false,
+}
+
+python_test_host {
+    name: "avf_early_vm_test",
+    main: "avf_early_vm_test.py",
+    srcs: ["avf_early_vm_test.py"],
+    device_first_data: [
+        ":avf_early_vm_test_kernel",
+        ":avf_early_vm_test_launcher",
+    ],
+    data: ["early_vms_rialto_test.xml"],
+    test_suites: ["general-tests"],
+    test_config: "AndroidTest.xml",
+    test_options: {
+        unit_test: false,
+    },
+}
diff --git a/tests/early_vm_test/AndroidTest.xml b/tests/early_vm_test/AndroidTest.xml
new file mode 100644
index 0000000..3eae96d
--- /dev/null
+++ b/tests/early_vm_test/AndroidTest.xml
@@ -0,0 +1,33 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!-- Copyright 2025 The Android Open Source Project
+
+     Licensed under the Apache License, Version 2.0 (the "License");
+     you may not use this file except in compliance with the License.
+     You may obtain a copy of the License at
+
+          http://www.apache.org/licenses/LICENSE-2.0
+
+     Unless required by applicable law or agreed to in writing, software
+     distributed under the License is distributed on an "AS IS" BASIS,
+     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+     See the License for the specific language governing permissions and
+     limitations under the License.
+-->
+<configuration description="Runs avf_early_vm_test.">
+    <target_preparer class="com.android.tradefed.targetprep.RootTargetPreparer">
+        <option name="force-root" value="true"/>
+    </target_preparer>
+
+    <target_preparer class="com.android.tradefed.targetprep.PushFilePreparer">
+    <option name="abort-on-push-failure" value="true" />
+        <option name="remount-system" value="true" />
+        <option name="push-file" key="avf_early_vm_test_launcher" value="/system_ext/bin/avf_early_vm_test_launcher" />
+        <option name="push-file" key="rialto.bin" value="/system_ext/etc/avf/rialto_test.bin" />
+        <option name="push-file" key="early_vms_rialto_test.xml" value="/system_ext/etc/avf/early_vms_rialto_test.xml" />
+    </target_preparer>
+
+    <test class="com.android.tradefed.testtype.python.PythonBinaryHostTest">
+        <option name="par-file-name" value="avf_early_vm_test" />
+        <option name="test-timeout" value="5m" />
+    </test>
+</configuration>
diff --git a/tests/early_vm_test/TEST_MAPPING b/tests/early_vm_test/TEST_MAPPING
new file mode 100644
index 0000000..1f2335b
--- /dev/null
+++ b/tests/early_vm_test/TEST_MAPPING
@@ -0,0 +1,9 @@
+// When adding or removing tests here, don't forget to amend _all_modules list in
+// wireless/android/busytown/ath_config/configs/prod/avf/tests.gcl
+{
+  "avf-presubmit": [
+    {
+      "name": "avf_early_vm_test"
+    }
+  ]
+}
diff --git a/tests/early_vm_test/avf_early_vm_test.py b/tests/early_vm_test/avf_early_vm_test.py
new file mode 100644
index 0000000..0003351
--- /dev/null
+++ b/tests/early_vm_test/avf_early_vm_test.py
@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+#
+# Copyright 2025 The Android Open Source Project
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import logging
+import os
+import subprocess
+import unittest
+
+_DEFAULT_COMMAND_TIMEOUT = 300
+_LAUNCHER_PATH = "/system_ext/bin/avf_early_vm_test_launcher"
+_RIALTO_PATH = "/system_ext/etc/avf/rialto_test.bin"
+
+def _RunCommand(cmd, timeout=_DEFAULT_COMMAND_TIMEOUT):
+    with subprocess.Popen(args=cmd,
+                          stderr=subprocess.PIPE,
+                          stdout=subprocess.PIPE,
+                          universal_newlines=True) as proc:
+        try:
+            out, err = proc.communicate(timeout=timeout)
+            returncode = proc.returncode
+        except subprocess.TimeoutExpired:
+            proc.kill()
+            out, err = proc.communicate()
+            returncode = proc.returncode
+
+    return out, err, returncode
+
+class AvfEarlyVmTest(unittest.TestCase):
+    def setUp(self):
+        self._serial_number = os.environ.get("ANDROID_SERIAL")
+        self.assertTrue(self._serial_number, "$ANDROID_SERIAL is empty.")
+
+    def _TestAvfEarlyVm(self, protected):
+        adb_cmd = ["adb", "-s", self._serial_number, "shell", _LAUNCHER_PATH, "--kernel",
+                   _RIALTO_PATH]
+        if protected:
+            adb_cmd.append("--protected")
+
+        _, err, returncode = _RunCommand(adb_cmd)
+        self.assertEqual(returncode, 0, f"{adb_cmd} failed: {err}")
+
+    def testAvfEarlyVmNonProtected(self):
+        self._TestAvfEarlyVm(False)
+
+    def testAvfEarlyVmProtected(self):
+        self._TestAvfEarlyVm(True)
+
+if __name__ == "__main__":
+    # Setting verbosity is required to generate output that the TradeFed test
+    # runner can parse.
+    unittest.main(verbosity=3)
diff --git a/tests/early_vm_test/early_vms_rialto_test.xml b/tests/early_vm_test/early_vms_rialto_test.xml
new file mode 100644
index 0000000..799fc3f
--- /dev/null
+++ b/tests/early_vm_test/early_vms_rialto_test.xml
@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!-- Copyright 2025 The Android Open Source Project
+
+     Licensed under the Apache License, Version 2.0 (the "License");
+     you may not use this file except in compliance with the License.
+     You may obtain a copy of the License at
+
+          http://www.apache.org/licenses/LICENSE-2.0
+
+     Unless required by applicable law or agreed to in writing, software
+     distributed under the License is distributed on an "AS IS" BASIS,
+     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+     See the License for the specific language governing permissions and
+     limitations under the License.
+-->
+<early_vms>
+    <early_vm>
+        <name>avf_early_vm_test_launcher</name>
+        <cid>299</cid>
+        <path>/system_ext/bin/avf_early_vm_test_launcher</path>
+    </early_vm>
+</early_vms>
diff --git a/tests/early_vm_test/src/main.rs b/tests/early_vm_test/src/main.rs
new file mode 100644
index 0000000..a3c80ca
--- /dev/null
+++ b/tests/early_vm_test/src/main.rs
@@ -0,0 +1,116 @@
+// Copyright 2025 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Tests running an early VM
+
+use android_system_virtualizationservice::{
+    aidl::android::system::virtualizationservice::{
+        IVirtualizationService::IVirtualizationService, VirtualMachineConfig::VirtualMachineConfig,
+        VirtualMachineRawConfig::VirtualMachineRawConfig,
+    },
+    binder::{ParcelFileDescriptor, ProcessState, Strong},
+};
+use anyhow::{Context, Result};
+use clap::Parser;
+use log::info;
+use std::fs::File;
+use std::path::PathBuf;
+
+use service_vm_comm::{Request, Response, VmType};
+use service_vm_manager::ServiceVm;
+use vmclient::VmInstance;
+
+const VM_MEMORY_MB: i32 = 16;
+
+#[derive(Parser)]
+/// Collection of CLI for avf_early_vm_test_rialto
+pub struct Args {
+    /// Path to the Rialto kernel image.
+    #[arg(long)]
+    kernel: PathBuf,
+
+    /// Whether the VM is protected or not.
+    #[arg(long)]
+    protected: bool,
+}
+
+fn get_service() -> Result<Strong<dyn IVirtualizationService>> {
+    let virtmgr = vmclient::VirtualizationService::new_early()
+        .context("Failed to spawn VirtualizationService")?;
+    virtmgr.connect().context("Failed to connect to VirtualizationService")
+}
+
+fn main() -> Result<()> {
+    if std::env::consts::ARCH != "aarch64" {
+        info!("{} not supported. skipping test", std::env::consts::ARCH);
+        return Ok(());
+    }
+
+    if !cfg!(early_vm_enabled) {
+        info!("early VM disabled. skipping test");
+        return Ok(());
+    }
+
+    let args = Args::parse();
+
+    if args.protected {
+        if !hypervisor_props::is_protected_vm_supported()? {
+            info!("pVMs are not supported on device. skipping test");
+            return Ok(());
+        }
+    } else if !hypervisor_props::is_vm_supported()? {
+        info!("non-pVMs are not supported on device. skipping test");
+        return Ok(());
+    }
+
+    let service = get_service()?;
+    let kernel =
+        File::open(&args.kernel).with_context(|| format!("Failed to open {:?}", &args.kernel))?;
+    let kernel = ParcelFileDescriptor::new(kernel);
+
+    let vm_config = VirtualMachineConfig::RawConfig(VirtualMachineRawConfig {
+        name: "avf_early_vm_test_launcher".to_owned(),
+        kernel: Some(kernel),
+        protectedVm: args.protected,
+        memoryMib: VM_MEMORY_MB,
+        platformVersion: "~1.0".to_owned(),
+        ..Default::default()
+    });
+
+    let vm_instance = VmInstance::create(
+        service.as_ref(),
+        &vm_config,
+        // console_in, console_out, and log will be redirected to the kernel log by virtmgr
+        None, // console_in
+        None, // console_out
+        None, // log
+        None, // dump_dt
+        None, // callback
+    )
+    .context("Failed to create VM")?;
+
+    ProcessState::start_thread_pool();
+
+    let vm_type = if args.protected { VmType::ProtectedVm } else { VmType::NonProtectedVm };
+    let mut vm_service = ServiceVm::start_vm(vm_instance, vm_type)?;
+
+    let request_data = vec![1, 2, 3, 4, 5];
+    let reversed_data = vec![5, 4, 3, 2, 1];
+    let response = vm_service
+        .process_request(Request::Reverse(request_data))
+        .context("Failed to process request")?;
+    assert_eq!(Response::Reverse(reversed_data), response);
+
+    Ok(())
+}
diff --git a/tests/helper/src/java/com/android/microdroid/test/device/MicrodroidDeviceTestBase.java b/tests/helper/src/java/com/android/microdroid/test/device/MicrodroidDeviceTestBase.java
index a932d32..c05fb0b 100644
--- a/tests/helper/src/java/com/android/microdroid/test/device/MicrodroidDeviceTestBase.java
+++ b/tests/helper/src/java/com/android/microdroid/test/device/MicrodroidDeviceTestBase.java
@@ -278,9 +278,11 @@
     }
 
     protected void assumeNoUpdatableVmSupport() throws VirtualMachineException {
-        assume().withMessage("Secretkeeper not supported")
-                .that(getVirtualMachineManager().isUpdatableVmSupported())
-                .isFalse();
+        assume().withMessage("Secretkeeper not supported").that(isUpdatableVmSupported()).isFalse();
+    }
+
+    protected boolean isUpdatableVmSupported() throws VirtualMachineException {
+        return getVirtualMachineManager().isUpdatableVmSupported();
     }
 
     protected void ensureVmAttestationSupported() throws Exception {
@@ -612,6 +614,8 @@
         public String mConsoleInput;
         public byte[] mInstanceSecret;
         public int mPageSize;
+        public byte[] mPayloadRpData;
+        public boolean mIsNewInstance;
 
         public void assertNoException() {
             if (mException != null) {
diff --git a/tests/testapk/src/java/com/android/microdroid/test/MicrodroidTests.java b/tests/testapk/src/java/com/android/microdroid/test/MicrodroidTests.java
index 6260458..a50ce98 100644
--- a/tests/testapk/src/java/com/android/microdroid/test/MicrodroidTests.java
+++ b/tests/testapk/src/java/com/android/microdroid/test/MicrodroidTests.java
@@ -1866,8 +1866,119 @@
         assertThat(testResults.mFileContent).isEqualTo(EXAMPLE_STRING);
     }
 
+    private boolean deviceCapableOfProtectedVm() {
+        int capabilities = getVirtualMachineManager().getCapabilities();
+        if ((capabilities & CAPABILITY_PROTECTED_VM) != 0) {
+            return true;
+        }
+        return false;
+    }
+
+    private void ensureUpdatableVmSupported() throws Exception {
+        if (getVendorApiLevel() >= 202504 && deviceCapableOfProtectedVm()) {
+            assertTrue(
+                    "Missing Updatable VM support, have you declared Secretkeeper interface?",
+                    isUpdatableVmSupported());
+        } else {
+            assumeTrue("Device does not support Updatable VM", isUpdatableVmSupported());
+        }
+    }
+
+    @Test
+    public void rollbackProtectedDataOfPayload() throws Exception {
+        assumeSupportedDevice();
+        // Rollback protected data is only possible if Updatable VMs is supported -
+        // which implies Secretkeeper support.
+        ensureUpdatableVmSupported();
+        byte[] value1 = new byte[32];
+        Arrays.fill(value1, (byte) 0xcc);
+        byte[] value2 = new byte[32];
+        Arrays.fill(value2, (byte) 0xdd);
+
+        VirtualMachineConfig config =
+                newVmConfigBuilderWithPayloadBinary("MicrodroidTestNativeLib.so")
+                        .setMemoryBytes(minMemoryRequired())
+                        .setEncryptedStorageBytes(ENCRYPTED_STORAGE_BYTES)
+                        .setDebugLevel(DEBUG_LEVEL_FULL)
+                        .build();
+        VirtualMachine vm = forceCreateNewVirtualMachine("test_vm", config);
+        TestResults testResults =
+                runVmTestService(
+                        TAG,
+                        vm,
+                        (ts, tr) -> {
+                            tr.mPayloadRpData = ts.insecurelyReadPayloadRpData();
+                        });
+        // ainsecurelyReadPayloadRpData()` must've failed since no data was ever written!
+        assertWithMessage("The read (unexpectedly) succeeded!")
+                .that(testResults.mException)
+                .isNotNull();
+
+        // Re-run the same VM & write/read th RP data & verify it what we just wrote!
+        testResults =
+                runVmTestService(
+                        TAG,
+                        vm,
+                        (ts, tr) -> {
+                            ts.insecurelyWritePayloadRpData(value1);
+                            tr.mPayloadRpData = ts.insecurelyReadPayloadRpData();
+                            ts.insecurelyWritePayloadRpData(value2);
+                        });
+        testResults.assertNoException();
+        assertThat(testResults.mPayloadRpData).isEqualTo(value1);
+
+        // Re-run the same VM again
+        testResults =
+                runVmTestService(
+                        TAG,
+                        vm,
+                        (ts, tr) -> {
+                            tr.mPayloadRpData = ts.insecurelyReadPayloadRpData();
+                        });
+        testResults.assertNoException();
+        assertThat(testResults.mPayloadRpData).isEqualTo(value2);
+    }
+
     @Test
     @CddTest
+    public void isNewInstanceTest() throws Exception {
+        assumeSupportedDevice();
+
+        VirtualMachineConfig config =
+                newVmConfigBuilderWithPayloadBinary("MicrodroidTestNativeLib.so")
+                        .setMemoryBytes(minMemoryRequired())
+                        .setDebugLevel(DEBUG_LEVEL_FULL)
+                        .build();
+        // TODO(b/325094712): Cuttlefish doesn't support device tree overlays which is required to
+        // find if the VM run is a new instance.
+        assumeFalse(
+                "Cuttlefish/Goldfish doesn't support device tree under /proc/device-tree",
+                isCuttlefish() || isGoldfish());
+        VirtualMachine vm = forceCreateNewVirtualMachine("test_vm_a", config);
+        TestResults testResults =
+                runVmTestService(
+                        TAG,
+                        vm,
+                        (ts, tr) -> {
+                            tr.mIsNewInstance = ts.isNewInstance();
+                        });
+        testResults.assertNoException();
+        assertThat(testResults.mIsNewInstance).isTrue();
+
+        // Re-run the same VM & ensure isNewInstance is false.
+        testResults =
+                runVmTestService(
+                        TAG,
+                        vm,
+                        (ts, tr) -> {
+                            tr.mIsNewInstance = ts.isNewInstance();
+                        });
+        testResults.assertNoException();
+        assertThat(testResults.mIsNewInstance).isFalse();
+    }
+
+    @Test
+    @CddTest(requirements = {"9.17/C-1-1", "9.17/C-2-1"})
     public void canReadFileFromAssets_debugFull() throws Exception {
         assumeSupportedDevice();
 
diff --git a/tests/testapk/src/native/testbinary.cpp b/tests/testapk/src/native/testbinary.cpp
index 7edabfd..06c7e9d 100644
--- a/tests/testapk/src/native/testbinary.cpp
+++ b/tests/testapk/src/native/testbinary.cpp
@@ -347,6 +347,28 @@
             return ScopedAStatus::ok();
         }
 
+        ScopedAStatus insecurelyReadPayloadRpData(std::array<uint8_t, 32>* out) override {
+            int32_t ret = AVmPayload_readRollbackProtectedSecret(out->data(), 32);
+            if (ret != 32) {
+                return ScopedAStatus::fromServiceSpecificError(ret);
+            }
+            return ScopedAStatus::ok();
+        }
+
+        ScopedAStatus insecurelyWritePayloadRpData(
+                const std::array<uint8_t, 32>& inputData) override {
+            int32_t ret = AVmPayload_writeRollbackProtectedSecret(inputData.data(), 32);
+            if (ret != 32) {
+                return ScopedAStatus::fromServiceSpecificError(ret);
+            }
+            return ScopedAStatus::ok();
+        }
+
+        ScopedAStatus isNewInstance(bool* is_new_instance_out) override {
+            *is_new_instance_out = AVmPayload_isNewInstance();
+            return ScopedAStatus::ok();
+        }
+
         ScopedAStatus quit() override { exit(0); }
     };
     auto testService = ndk::SharedRefBase::make<TestService>();
diff --git a/tests/testapk/src/native/testbinary.rs b/tests/testapk/src/native/testbinary.rs
index a84b955..c9d46b8 100644
--- a/tests/testapk/src/native/testbinary.rs
+++ b/tests/testapk/src/native/testbinary.rs
@@ -126,6 +126,15 @@
     fn readLineFromConsole(&self) -> BinderResult<String> {
         unimplemented()
     }
+    fn insecurelyReadPayloadRpData(&self) -> BinderResult<[u8; 32]> {
+        unimplemented()
+    }
+    fn insecurelyWritePayloadRpData(&self, _: &[u8; 32]) -> BinderResult<()> {
+        unimplemented()
+    }
+    fn isNewInstance(&self) -> BinderResult<bool> {
+        unimplemented()
+    }
 }
 
 fn unimplemented<T>() -> BinderResult<T> {
diff --git a/tests/vmshareapp/src/java/com/android/microdroid/test/sharevm/VmShareServiceImpl.java b/tests/vmshareapp/src/java/com/android/microdroid/test/sharevm/VmShareServiceImpl.java
index 13b0c51..1f71888 100644
--- a/tests/vmshareapp/src/java/com/android/microdroid/test/sharevm/VmShareServiceImpl.java
+++ b/tests/vmshareapp/src/java/com/android/microdroid/test/sharevm/VmShareServiceImpl.java
@@ -276,5 +276,20 @@
         public void quit() throws RemoteException {
             throw new UnsupportedOperationException("Not supported");
         }
+
+        @Override
+        public byte[] insecurelyReadPayloadRpData() {
+            throw new UnsupportedOperationException("Not supported");
+        }
+
+        @Override
+        public void insecurelyWritePayloadRpData(byte[] data) {
+            throw new UnsupportedOperationException("Not supported");
+        }
+
+        @Override
+        public boolean isNewInstance() {
+            throw new UnsupportedOperationException("Not supported");
+        }
     }
 }