diff --git a/crates/sui-adapter/src/adapter.rs b/crates/sui-adapter/src/adapter.rs
index 62654c3d6f4ea..fad9773561409 100644
--- a/crates/sui-adapter/src/adapter.rs
+++ b/crates/sui-adapter/src/adapter.rs
@@ -35,7 +35,7 @@ use sui_types::{
     id::UID,
     messages::{CallArg, EntryArgumentErrorKind, InputObjectKind, ObjectArg},
     object::{self, Data, MoveObject, Object, Owner, ID_END_INDEX},
-    storage::{DeleteKind, ObjectChange, ParentSync, Storage},
+    storage::{DeleteKind, ObjectChange, ParentSync, Storage, WriteKind},
     SUI_SYSTEM_STATE_OBJECT_ID,
 };
 use sui_verifier::{
@@ -306,8 +306,7 @@ pub fn store_package_and_init_modules<
     // The call to unwrap() will go away once we remove address owner from Immutable objects.
     let package_object = Object::new_package(modules, ctx.digest());
     let id = package_object.id();
-    state_view.set_create_object_ids(BTreeSet::from([id]));
-    let changes = BTreeMap::from([(id, ObjectChange::Write(package_object))]);
+    let changes = BTreeMap::from([(id, ObjectChange::Write(package_object, WriteKind::Create))]);
     state_view.apply_object_changes(changes);
 
     init_modules(state_view, vm, modules_to_init, ctx, gas_status)
@@ -475,7 +474,7 @@ fn process_successful_execution<
             .try_as_move_mut()
             .expect("We previously checked that mutable ref inputs are Move objects")
             .update_contents_and_increment_version(new_contents);
-        changes.insert(obj_id, ObjectChange::Write(obj));
+        changes.insert(obj_id, ObjectChange::Write(obj, WriteKind::Mutate));
     }
     let tx_digest = ctx.digest();
     // newly_generated_ids contains all object IDs generated in this transaction.
@@ -509,7 +508,7 @@ fn process_successful_execution<
                     EventType::ShareObject => Owner::Shared,
                     _ => unreachable!(),
                 };
-                let obj = handle_transfer(
+                let (obj, write_kind) = handle_transfer(
                     ctx.sender(),
                     new_owner,
                     type_,
@@ -525,7 +524,7 @@ fn process_successful_execution<
                     &mut frozen_object_ids,
                     &mut child_count_deltas,
                 )?;
-                changes.insert(obj.id(), ObjectChange::Write(obj));
+                changes.insert(obj.id(), ObjectChange::Write(obj, write_kind));
             }
             EventType::DeleteObjectID => {
                 // unwrap safe because this event can only be emitted from processing
@@ -673,10 +672,10 @@ fn process_successful_execution<
             let mut object = state_view.read_object(&id).unwrap().clone();
             // Active input object must be Move object.
             object.data.try_as_move_mut().unwrap().increment_version();
-            ObjectChange::Write(object)
+            ObjectChange::Write(object, WriteKind::Mutate)
         });
         match change {
-            ObjectChange::Write(object) => {
+            ObjectChange::Write(object, _) => {
                 object
                     .data
                     .try_as_move_mut()
@@ -697,7 +696,6 @@ fn process_successful_execution<
     }
 
     // apply object writes and object deletions
-    state_view.set_create_object_ids(newly_generated_ids);
     state_view.apply_object_changes(changes);
 
     // check all frozen objects have a child count of zero
@@ -738,7 +736,7 @@ fn handle_transfer<
     newly_generated_unused: &mut BTreeSet<ObjectID>,
     frozen_object_ids: &mut BTreeSet<ObjectID>,
     child_count_deltas: &mut BTreeMap<ObjectID, i64>,
-) -> Result<Object, ExecutionError> {
+) -> Result<(Object, WriteKind), ExecutionError> {
     let s_type = match type_ {
         TypeTag::Struct(s_type) => s_type,
         _ => unreachable!("Only structs can be transferred"),
@@ -750,15 +748,15 @@ fn handle_transfer<
     newly_generated_unused.remove(&id);
     let old_object = by_value_objects.remove(&id);
     let mut is_unwrapped = !(newly_generated_ids.contains(&id) || id == SUI_SYSTEM_STATE_OBJECT_ID);
-    let (version, child_count) = match old_object {
-        Some((_, version, child_count)) => (version, child_count),
+    let (write_kind, version, child_count) = match old_object {
+        Some((_, version, child_count)) => (WriteKind::Mutate, version, child_count),
         // When an object was wrapped at version `v`, we added an record into `parent_sync`
         // with version `v+1` along with OBJECT_DIGEST_WRAPPED. Now when the object is unwrapped,
         // it will also have version `v+1`, leading to a violation of the invariant that any
         // object_id and version pair must be unique. We use the version from parent_sync and
         // increment it (below), so we will have `(v+1)+1`, thus preserving the uniqueness
         None if is_unwrapped => match state_view.get_latest_parent_entry_ref(id) {
-            Ok(Some((_, last_version, _))) => (last_version, None),
+            Ok(Some((_, last_version, _))) => (WriteKind::Unwrap, last_version, None),
             // if the object is not in parent sync, it was wrapped before ever being stored into
             // storage.
             // we set is_unwrapped to false since the object has never been in storage
@@ -766,7 +764,7 @@ fn handle_transfer<
             Ok(None) => {
                 is_unwrapped = false;
                 newly_generated_ids.insert(id);
-                (SequenceNumber::new(), None)
+                (WriteKind::Create, SequenceNumber::new(), None)
             }
             Err(_) => {
                 // TODO this error is (hopefully) transient and should not be
@@ -777,7 +775,7 @@ fn handle_transfer<
                 ));
             }
         },
-        None => (SequenceNumber::new(), None),
+        None => (WriteKind::Create, SequenceNumber::new(), None),
     };
 
     // safe because `has_public_transfer` was properly determined from the abilities
@@ -878,8 +876,7 @@ fn handle_transfer<
             object_owner_map.insert(obj_address, new_owner);
         }
     }
-
-    Ok(obj)
+    Ok((obj, write_kind))
 }
 
 #[cfg(debug_assertions)]
diff --git a/crates/sui-adapter/src/in_memory_storage.rs b/crates/sui-adapter/src/in_memory_storage.rs
index fc60ff47943fd..61debdcc4e470 100644
--- a/crates/sui-adapter/src/in_memory_storage.rs
+++ b/crates/sui-adapter/src/in_memory_storage.rs
@@ -7,7 +7,7 @@ use sui_types::{
     base_types::{ObjectID, ObjectRef, SequenceNumber},
     error::{SuiError, SuiResult},
     object::Object,
-    storage::{BackingPackageStore, DeleteKind, ParentSync},
+    storage::{BackingPackageStore, DeleteKind, ParentSync, WriteKind},
 };
 
 // TODO: We should use AuthorityTemporaryStore instead.
@@ -97,11 +97,11 @@ impl InMemoryStorage {
 
     pub fn finish(
         &mut self,
-        written: BTreeMap<ObjectID, (ObjectRef, Object)>,
+        written: BTreeMap<ObjectID, (ObjectRef, Object, WriteKind)>,
         deleted: BTreeMap<ObjectID, (SequenceNumber, DeleteKind)>,
     ) {
         debug_assert!(written.keys().all(|id| !deleted.contains_key(id)));
-        for (_id, (_, new_object)) in written {
+        for (_id, (_, new_object, _)) in written {
             debug_assert!(new_object.id() == _id);
             self.insert_object(new_object);
         }
diff --git a/crates/sui-adapter/src/temporary_store.rs b/crates/sui-adapter/src/temporary_store.rs
index 6e15dde81b8ef..3230c51e4be31 100644
--- a/crates/sui-adapter/src/temporary_store.rs
+++ b/crates/sui-adapter/src/temporary_store.rs
@@ -4,7 +4,7 @@
 use move_core_types::account_address::AccountAddress;
 use move_core_types::language_storage::{ModuleId, StructTag};
 use move_core_types::resolver::{ModuleResolver, ResourceResolver};
-use std::collections::{BTreeMap, BTreeSet};
+use std::collections::BTreeMap;
 use sui_types::base_types::{
     ObjectDigest, ObjectID, ObjectRef, SequenceNumber, SuiAddress, TransactionDigest,
 };
@@ -12,7 +12,9 @@ use sui_types::error::{ExecutionError, SuiError, SuiResult};
 use sui_types::fp_bail;
 use sui_types::messages::{ExecutionStatus, InputObjects, TransactionEffects};
 use sui_types::object::{Data, Object};
-use sui_types::storage::{BackingPackageStore, DeleteKind, ObjectChange, ParentSync, Storage};
+use sui_types::storage::{
+    BackingPackageStore, DeleteKind, ObjectChange, ParentSync, Storage, WriteKind,
+};
 use sui_types::{
     event::Event,
     gas::{GasCostSummary, SuiGasStatus},
@@ -22,7 +24,7 @@ use sui_types::{
 pub struct InnerTemporaryStore {
     pub objects: BTreeMap<ObjectID, Object>,
     pub mutable_inputs: Vec<ObjectRef>,
-    pub written: BTreeMap<ObjectID, (ObjectRef, Object)>,
+    pub written: BTreeMap<ObjectID, (ObjectRef, Object, WriteKind)>,
     pub deleted: BTreeMap<ObjectID, (SequenceNumber, DeleteKind)>,
 }
 
@@ -39,14 +41,11 @@ pub struct TemporaryStore<S> {
     // When an object is being written, we need to ensure that a few invariants hold.
     // It's critical that we always call write_object to update `_written`, instead of writing
     // into _written directly.
-    _written: BTreeMap<ObjectID, Object>, // Objects written
+    _written: BTreeMap<ObjectID, (Object, WriteKind)>, // Objects written
     /// Objects actively deleted.
     deleted: BTreeMap<ObjectID, (SequenceNumber, DeleteKind)>,
     /// Ordered sequence of events emitted by execution
     events: Vec<Event>,
-    // New object IDs created during the transaction, needed for
-    // telling apart unwrapped objects.
-    created_object_ids: BTreeSet<ObjectID>,
 }
 
 impl<S> TemporaryStore<S> {
@@ -63,7 +62,6 @@ impl<S> TemporaryStore<S> {
             _written: BTreeMap::new(),
             deleted: BTreeMap::new(),
             events: Vec::new(),
-            created_object_ids: BTreeSet::new(),
         }
     }
 
@@ -85,7 +83,7 @@ impl<S> TemporaryStore<S> {
         let written = self
             ._written
             .into_iter()
-            .map(|(id, obj)| (id, (obj.compute_object_reference(), obj)))
+            .map(|(id, (obj, kind))| (id, (obj.compute_object_reference(), obj, kind)))
             .collect();
         let deleted = self
             .deleted
@@ -124,7 +122,8 @@ impl<S> TemporaryStore<S> {
             }
         }
         for object in to_be_updated {
-            self.write_object(object);
+            // The object must be mutated as it was present in the input objects
+            self.write_object(object, WriteKind::Mutate);
         }
     }
 
@@ -144,18 +143,14 @@ impl<S> TemporaryStore<S> {
             gas_object_size,
             gas_object.storage_rebate.into(),
         )?;
-        objects_to_update.push(gas_object.clone());
-
-        for (object_id, object) in &mut self._written {
-            let (old_object_size, storage_rebate) =
-                if let Some(old_object) = self.input_objects.get(object_id) {
-                    (
-                        old_object.object_size_for_gas_metering(),
-                        old_object.storage_rebate,
-                    )
-                } else {
-                    (0, 0)
-                };
+        objects_to_update.push((gas_object.clone(), WriteKind::Mutate));
+
+        for (object_id, (object, write_kind)) in &mut self._written {
+            let (old_object_size, storage_rebate) = self
+                .input_objects
+                .get(object_id)
+                .map(|old| (old.object_size_for_gas_metering(), old.storage_rebate))
+                .unwrap_or((0, 0));
             let new_storage_rebate = gas_status.charge_storage_mutation(
                 old_object_size,
                 object.object_size_for_gas_metering(),
@@ -165,7 +160,7 @@ impl<S> TemporaryStore<S> {
                 // We don't need to set storage rebate for immutable objects, as they will
                 // never be deleted.
                 object.storage_rebate = new_storage_rebate;
-                objects_to_update.push(object.clone());
+                objects_to_update.push((object.clone(), *write_kind));
             }
         }
 
@@ -185,8 +180,8 @@ impl<S> TemporaryStore<S> {
 
         // Write all objects at the end only if all previous gas charges succeeded.
         // This avoids polluting the temporary store state if this function failed.
-        for object in objects_to_update {
-            self.write_object(object);
+        for (object, write_kind) in objects_to_update {
+            self.write_object(object, write_kind);
         }
 
         Ok(())
@@ -201,35 +196,31 @@ impl<S> TemporaryStore<S> {
         status: ExecutionStatus,
         gas_object_ref: ObjectRef,
     ) -> (InnerTemporaryStore, TransactionEffects) {
-        let written = self
+        let written: BTreeMap<ObjectID, (ObjectRef, Owner, WriteKind)> = self
             ._written
             .iter()
-            .map(|(id, obj)| (*id, (obj.compute_object_reference(), obj.owner)))
-            .collect::<BTreeMap<_, _>>();
+            .map(|(id, (obj, write_kind))| {
+                let obj_ref = obj.compute_object_reference();
+                (*id, (obj_ref, obj.owner, *write_kind))
+            })
+            .collect();
 
         // In the case of special transactions that don't require a gas object,
         // we don't really care about the effects to gas, just use the input for it.
         let updated_gas_object_info = if gas_object_ref.0 == ObjectID::ZERO {
             (gas_object_ref, Owner::AddressOwner(SuiAddress::default()))
         } else {
-            written[&gas_object_ref.0]
+            let (obj_ref, owner, _kind) = written[&gas_object_ref.0];
+            (obj_ref, owner)
         };
-        let mut created = vec![];
         let mut mutated = vec![];
+        let mut created = vec![];
         let mut unwrapped = vec![];
-        for (id, object_ref_and_owner) in written {
-            match (
-                self.created_object_ids.contains(&id),
-                self.input_objects.contains_key(&id),
-            ) {
-                (true, _) => created.push(object_ref_and_owner),
-                (false, true) => mutated.push(object_ref_and_owner),
-                (false, false) => {
-                    // wrapped objects must have their version set to 1 + the last known version in
-                    // the `parent_sync`
-                    debug_assert!(object_ref_and_owner.0 .1.value() > 1);
-                    unwrapped.push(object_ref_and_owner)
-                }
+        for (_id, (object_ref, owner, kind)) in written {
+            match kind {
+                WriteKind::Mutate => mutated.push((object_ref, owner)),
+                WriteKind::Create => created.push((object_ref, owner)),
+                WriteKind::Unwrap => unwrapped.push((object_ref, owner)),
             }
         }
 
@@ -292,19 +283,11 @@ impl<S> TemporaryStore<S> {
             "Mutable input neither written nor deleted."
         );
 
-        debug_assert!(
-            {
-                let input_ids = self.input_objects.clone().into_keys().collect();
-                self.created_object_ids.is_disjoint(&input_ids)
-            },
-            "Newly created object IDs showed up in the input",
-        );
-
         debug_assert!(
             {
                 self._written
                     .iter()
-                    .all(|(_, obj)| obj.previous_transaction == self.tx_digest)
+                    .all(|(_, (obj, _))| obj.previous_transaction == self.tx_digest)
             },
             "Object previous transaction not properly set",
         );
@@ -314,7 +297,7 @@ impl<S> TemporaryStore<S> {
     // is that an entry is not both added and deleted by the
     // caller.
 
-    pub fn write_object(&mut self, mut object: Object) {
+    pub fn write_object(&mut self, mut object: Object, kind: WriteKind) {
         // there should be no write after delete
         debug_assert!(self.deleted.get(&object.id()) == None);
         // Check it is not read-only
@@ -330,7 +313,7 @@ impl<S> TemporaryStore<S> {
         // The adapter is not very disciplined at filling in the correct
         // previous transaction digest, so we ensure it is correct here.
         object.previous_transaction = self.tx_digest;
-        self._written.insert(object.id(), object);
+        self._written.insert(object.id(), (object, kind));
     }
 
     pub fn delete_object(&mut self, id: &ObjectID, version: SequenceNumber, kind: DeleteKind) {
@@ -358,26 +341,25 @@ impl<S> Storage for TemporaryStore<S> {
         self._written.clear();
         self.deleted.clear();
         self.events.clear();
-        self.created_object_ids.clear();
     }
 
     fn read_object(&self, id: &ObjectID) -> Option<&Object> {
         // there should be no read after delete
         debug_assert!(self.deleted.get(id) == None);
-        self._written.get(id).or_else(|| self.input_objects.get(id))
-    }
-
-    fn set_create_object_ids(&mut self, ids: BTreeSet<ObjectID>) {
-        self.created_object_ids = ids;
+        self._written
+            .get(id)
+            .map(|(obj, _kind)| obj)
+            .or_else(|| self.input_objects.get(id))
     }
 
     fn log_event(&mut self, event: Event) {
         self.events.push(event)
     }
+
     fn apply_object_changes(&mut self, changes: BTreeMap<ObjectID, ObjectChange>) {
         for (id, change) in changes {
             match change {
-                ObjectChange::Write(new_value) => self.write_object(new_value),
+                ObjectChange::Write(new_value, kind) => self.write_object(new_value, kind),
                 ObjectChange::Delete(version, kind) => self.delete_object(&id, version, kind),
             }
         }
diff --git a/crates/sui-core/src/authority.rs b/crates/sui-core/src/authority.rs
index c59c3e5df6a06..b09e1815b6da9 100644
--- a/crates/sui-core/src/authority.rs
+++ b/crates/sui-core/src/authority.rs
@@ -866,7 +866,10 @@ impl AuthorityState {
                 .input_objects()?
                 .iter()
                 .map(|o| o.object_id()),
-            effects.effects.all_mutated(),
+            effects
+                .effects
+                .all_mutated()
+                .map(|(obj_ref, owner, _kind)| (*obj_ref, *owner)),
             cert.signed_data
                 .data
                 .move_calls()
diff --git a/crates/sui-core/src/authority/authority_store.rs b/crates/sui-core/src/authority/authority_store.rs
index c4a793aba38da..5ae55bad01c00 100644
--- a/crates/sui-core/src/authority/authority_store.rs
+++ b/crates/sui-core/src/authority/authority_store.rs
@@ -6,6 +6,7 @@ use narwhal_executor::ExecutionIndices;
 use rocksdb::Options;
 use serde::{Deserialize, Serialize};
 use serde_with::serde_as;
+use std::collections::BTreeMap;
 use std::iter;
 use std::path::Path;
 use std::sync::atomic::AtomicU64;
@@ -18,6 +19,7 @@ use sui_storage::{
 use sui_types::batch::{SignedBatch, TxSequenceNumber};
 use sui_types::crypto::{AuthoritySignInfo, EmptySignInfo};
 use sui_types::object::{Owner, OBJECT_START_VERSION};
+use sui_types::storage::WriteKind;
 use sui_types::{base_types::SequenceNumber, storage::ParentSync};
 use tokio::sync::Notify;
 use tokio_retry::strategy::{jitter, ExponentialBackoff};
@@ -674,7 +676,7 @@ impl<S: Eq + Debug + Serialize + for<'de> Deserialize<'de>> SuiDataStore<S> {
     pub async fn update_gateway_state(
         &self,
         input_objects: InputObjects,
-        mutated_objects: HashMap<ObjectRef, Object>,
+        mutated_objects: BTreeMap<ObjectRef, (Object, WriteKind)>,
         certificate: CertifiedTransaction,
         proposed_seq: TxSequenceNumber,
         effects: TransactionEffectsEnvelope<S>,
@@ -683,8 +685,8 @@ impl<S: Eq + Debug + Serialize + for<'de> Deserialize<'de>> SuiDataStore<S> {
         let transaction_digest = certificate.digest();
         let mut temporary_store =
             TemporaryStore::new(Arc::new(&self), input_objects, *transaction_digest);
-        for (_, object) in mutated_objects {
-            temporary_store.write_object(object);
+        for (_, (object, kind)) in mutated_objects {
+            temporary_store.write_object(object, kind);
         }
         for obj_ref in &effects.effects.deleted {
             temporary_store.delete_object(&obj_ref.0, obj_ref.1, DeleteKind::Normal);
@@ -692,16 +694,14 @@ impl<S: Eq + Debug + Serialize + for<'de> Deserialize<'de>> SuiDataStore<S> {
         for obj_ref in &effects.effects.wrapped {
             temporary_store.delete_object(&obj_ref.0, obj_ref.1, DeleteKind::Wrap);
         }
+        let (inner_temporary_store, _events) = temporary_store.into_inner();
 
         let mut write_batch = self.tables.certificates.batch();
-
         // Store the certificate indexed by transaction digest
         write_batch = write_batch.insert_batch(
             &self.tables.certificates,
             std::iter::once((transaction_digest, &certificate)),
         )?;
-
-        let (inner_temporary_store, _events) = temporary_store.into_inner();
         self.sequence_tx(
             write_batch,
             inner_temporary_store,
@@ -788,7 +788,7 @@ impl<S: Eq + Debug + Serialize + for<'de> Deserialize<'de>> SuiDataStore<S> {
             written,
             deleted,
         } = inner_temporary_store;
-        trace!(written =? written.values().map(|((obj_id, ver, _), _)| (obj_id, ver)).collect::<Vec<_>>(),
+        trace!(written =? written.values().map(|((obj_id, ver, _), _, _)| (obj_id, ver)).collect::<Vec<_>>(),
                "batch_update_objects: temp store written");
 
         let owned_inputs: Vec<_> = active_inputs
@@ -802,22 +802,21 @@ impl<S: Eq + Debug + Serialize + for<'de> Deserialize<'de>> SuiDataStore<S> {
         // For wrapped objects, although their owners technically didn't change, we will lose track
         // of them and there is no guarantee on their owner in the future. Hence we treat them
         // the same as deleted.
-        let old_object_owners = deleted
-            .iter()
-            // We need to call get() on objects because some object that were just deleted may not
-            // be in the objects list. This can happen if these deleted objects were wrapped in the past,
-            // and hence will not show up in the input objects.
-            .filter_map(|(id, _)| objects.get(id).and_then(Object::get_owner_and_id))
-            .chain(
-                written
-                    .iter()
-                    .filter_map(|(id, (_, new_object))| match objects.get(id) {
+        let old_object_owners =
+            deleted
+                .iter()
+                // We need to call get() on objects because some object that were just deleted may not
+                // be in the objects list. This can happen if these deleted objects were wrapped in the past,
+                // and hence will not show up in the input objects.
+                .filter_map(|(id, _)| objects.get(id).and_then(Object::get_owner_and_id))
+                .chain(written.iter().filter_map(
+                    |(id, (_, new_object, _))| match objects.get(id) {
                         Some(old_object) if old_object.owner != new_object.owner => {
                             old_object.get_owner_and_id()
                         }
                         _ => None,
-                    }),
-            );
+                    },
+                ));
 
         // Delete the old owner index entries
         write_batch = write_batch.delete_batch(&self.tables.owner_index, old_object_owners)?;
@@ -827,7 +826,7 @@ impl<S: Eq + Debug + Serialize + for<'de> Deserialize<'de>> SuiDataStore<S> {
             &self.tables.parent_sync,
             written
                 .iter()
-                .map(|(_, (object_ref, _object_))| (object_ref, transaction_digest)),
+                .map(|(_, (object_ref, _object, _kind))| (object_ref, transaction_digest)),
         )?;
 
         // Index the certificate by the objects deleted
@@ -862,7 +861,7 @@ impl<S: Eq + Debug + Serialize + for<'de> Deserialize<'de>> SuiDataStore<S> {
             &self.tables.owner_index,
             written
                 .iter()
-                .filter_map(|(_id, (object_ref, new_object))| {
+                .filter_map(|(_id, (object_ref, new_object, _kind))| {
                     trace!(?object_ref, owner =? new_object.owner, "Updating owner_index");
                     new_object
                         .get_owner_and_id()
@@ -875,7 +874,7 @@ impl<S: Eq + Debug + Serialize + for<'de> Deserialize<'de>> SuiDataStore<S> {
             &self.tables.objects,
             written
                 .iter()
-                .map(|(_, (obj_ref, new_object))| (ObjectKey::from(obj_ref), new_object)),
+                .map(|(_, (obj_ref, new_object, _kind))| (ObjectKey::from(obj_ref), new_object)),
         )?;
 
         // Atomic write of all data other than locks
@@ -899,7 +898,7 @@ impl<S: Eq + Debug + Serialize + for<'de> Deserialize<'de>> SuiDataStore<S> {
             //    been deleted by a concurrent tx that finished first. In that case, check if the tx effects exist.
             let new_locks_to_init: Vec<_> = written
                 .iter()
-                .filter_map(|(_, (object_ref, new_object))| {
+                .filter_map(|(_, (object_ref, new_object, _kind))| {
                     if new_object.is_owned_or_quasi_shared() {
                         Some(*object_ref)
                     } else {
diff --git a/crates/sui-core/src/execution_engine.rs b/crates/sui-core/src/execution_engine.rs
index 8be79c7112b3c..8432f7271317a 100644
--- a/crates/sui-core/src/execution_engine.rs
+++ b/crates/sui-core/src/execution_engine.rs
@@ -5,7 +5,7 @@ use move_core_types::ident_str;
 use move_core_types::identifier::Identifier;
 use std::{collections::BTreeSet, sync::Arc};
 use sui_adapter::temporary_store::InnerTemporaryStore;
-use sui_types::storage::ParentSync;
+use sui_types::storage::{ParentSync, WriteKind};
 
 use crate::authority::TemporaryStore;
 use move_core_types::language_storage::ModuleId;
@@ -246,7 +246,7 @@ fn execute_transaction<S: BackingPackageStore + ParentSync>(
         gas_object = temporary_store.read_object(&gas_object_id).unwrap().clone();
         gas::deduct_gas(&mut gas_object, gas_used, gas_rebate);
         trace!(gas_used, gas_obj_id =? gas_object.id(), gas_obj_ver =? gas_object.version(), "Updated gas object");
-        temporary_store.write_object(gas_object);
+        temporary_store.write_object(gas_object, WriteKind::Mutate);
     }
 
     let cost_summary = gas_status.summary(result.is_ok());
@@ -273,7 +273,7 @@ fn transfer_object<S>(
         type_: TransferType::Coin,
         amount,
     });
-    temporary_store.write_object(object);
+    temporary_store.write_object(object, WriteKind::Mutate);
     Ok(())
 }
 
@@ -318,11 +318,7 @@ fn transfer_sui<S>(
             Owner::AddressOwner(recipient),
             tx_ctx.digest(),
         );
-        temporary_store.write_object(new_object);
-
-        // This is necessary for the temporary store to know this new object is not unwrapped.
-        let newly_generated_ids = tx_ctx.recreate_all_ids();
-        temporary_store.set_create_object_ids(newly_generated_ids);
+        temporary_store.write_object(new_object, WriteKind::Create);
         Some(amount)
     } else {
         // If amount is not specified, we simply transfer the entire coin object.
@@ -345,7 +341,7 @@ fn transfer_sui<S>(
     #[cfg(debug_assertions)]
     assert_eq!(object.version(), version);
 
-    temporary_store.write_object(object);
+    temporary_store.write_object(object, WriteKind::Mutate);
 
     Ok(())
 }
diff --git a/crates/sui-core/src/gateway_state.rs b/crates/sui-core/src/gateway_state.rs
index 49571285e7747..3e4dfe68a3fe5 100644
--- a/crates/sui-core/src/gateway_state.rs
+++ b/crates/sui-core/src/gateway_state.rs
@@ -3,7 +3,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use std::cmp::Ordering;
-use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
+use std::collections::{BTreeMap, BTreeSet, HashSet};
 use std::path::Path;
 use std::sync::atomic::AtomicU64;
 use std::sync::Arc;
@@ -653,21 +653,29 @@ where
         );
 
         // Download the latest content of every mutated object from the authorities.
-        let mutated_object_refs: BTreeSet<_> = effects
-            .effects
-            .all_mutated()
-            .map(|(obj_ref, _)| *obj_ref)
-            .collect();
+        let mut mutated_object_kinds = BTreeMap::new();
+        let mut mutated_object_refs = BTreeSet::new();
+        for (obj_ref, _, kind) in effects.effects.all_mutated() {
+            mutated_object_kinds.insert(obj_ref.0, kind);
+            mutated_object_refs.insert(*obj_ref);
+        }
         let mutated_objects = self
             .download_objects_from_authorities(mutated_object_refs)
             .await?;
+        let mutated_objects_with_kind = mutated_objects
+            .into_iter()
+            .map(|(obj_ref, obj)| {
+                let kind = mutated_object_kinds.get(&obj_ref.0).copied().unwrap();
+                (obj_ref, (obj, kind))
+            })
+            .collect();
         let seq = self
             .next_tx_seq_number
             .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
         self.store
             .update_gateway_state(
                 input_objects,
-                mutated_objects,
+                mutated_objects_with_kind,
                 new_certificate.clone(),
                 seq,
                 effects.clone().to_unsigned_effects(),
@@ -820,12 +828,12 @@ where
         &self,
         // TODO: HashSet probably works here just fine.
         object_refs: BTreeSet<ObjectRef>,
-    ) -> Result<HashMap<ObjectRef, Object>, SuiError> {
+    ) -> Result<BTreeMap<ObjectRef, Object>, SuiError> {
         let mut receiver = self
             .authorities
             .fetch_objects_from_authorities(object_refs.clone());
 
-        let mut objects = HashMap::new();
+        let mut objects = BTreeMap::new();
         while let Some(resp) = receiver.recv().await {
             if let Ok(o) = resp {
                 // TODO: Make fetch_objects_from_authorities also return object ref
@@ -913,13 +921,13 @@ where
         let mutated_objects = self.store.get_objects(
             &effects
                 .all_mutated()
-                .map(|((object_id, _, _), _)| *object_id)
+                .map(|((object_id, _, _), _, _)| *object_id)
                 .collect::<Vec<_>>(),
         )?;
         let mut updated_gas = None;
         let mut package = None;
         let mut created_objects = vec![];
-        for ((obj_ref, _), object) in effects.all_mutated().zip(mutated_objects) {
+        for ((obj_ref, _, _), object) in effects.all_mutated().zip(mutated_objects) {
             let object = object.ok_or(SuiError::InconsistentGatewayResult {
                 error: format!(
                     "Crated/Updated object doesn't exist in the store: {:?}",
diff --git a/crates/sui-storage/src/indexes.rs b/crates/sui-storage/src/indexes.rs
index a3dc94595ccce..4d6bbacdc48c4 100644
--- a/crates/sui-storage/src/indexes.rs
+++ b/crates/sui-storage/src/indexes.rs
@@ -74,11 +74,11 @@ fn timestamps_table_default_config() -> Options {
 }
 
 impl IndexStore {
-    pub fn index_tx<'a>(
+    pub fn index_tx(
         &self,
         sender: SuiAddress,
         active_inputs: impl Iterator<Item = ObjectID>,
-        mutated_objects: impl Iterator<Item = &'a (ObjectRef, Owner)> + Clone,
+        mutated_objects: impl Iterator<Item = (ObjectRef, Owner)> + Clone,
         move_functions: impl Iterator<Item = (ObjectID, Identifier, Identifier)> + Clone,
         sequence: TxSequenceNumber,
         digest: &TransactionDigest,
diff --git a/crates/sui-transactional-test-runner/src/test_adapter.rs b/crates/sui-transactional-test-runner/src/test_adapter.rs
index 79ce8549056bf..bd0be4d8dd742 100644
--- a/crates/sui-transactional-test-runner/src/test_adapter.rs
+++ b/crates/sui-transactional-test-runner/src/test_adapter.rs
@@ -36,9 +36,9 @@ use std::{
     path::Path,
     sync::Arc,
 };
+use sui_adapter::in_memory_storage::InMemoryStorage;
 use sui_adapter::temporary_store::TemporaryStore;
 use sui_adapter::{adapter::new_move_vm, genesis};
-use sui_adapter::{in_memory_storage::InMemoryStorage, temporary_store::InnerTemporaryStore};
 use sui_core::execution_engine;
 use sui_framework::DEFAULT_FRAMEWORK_PATH;
 use sui_types::{
@@ -489,12 +489,12 @@ impl<'a> SuiTestAdapter<'a> {
                 status,
                 events,
                 created,
+                mutated,
+                unwrapped,
                 // TODO display all these somehow
                 transaction_digest: _,
-                mutated: _,
-                unwrapped: _,
-                deleted: _,
-                wrapped: _,
+                deleted,
+                wrapped,
                 gas_object: _,
                 ..
             },
@@ -511,29 +511,27 @@ impl<'a> SuiTestAdapter<'a> {
             // TODO: Support different epochs in transactional tests.
             0,
         );
-        let InnerTemporaryStore {
-            written, deleted, ..
-        } = inner;
         let created_set: BTreeSet<_> = created.iter().map(|((id, _, _), _)| *id).collect();
         let mut created_ids: Vec<_> = created_set.iter().copied().collect();
-        let mut written_ids: Vec<_> = written
-            .keys()
-            .copied()
-            .filter(|id| !created_set.contains(id))
+        let mut written_ids: Vec<_> = mutated
+            .iter()
+            .chain(&unwrapped)
+            .map(|((id, _, _), _)| *id)
+            .collect();
+        let mut deleted_ids: Vec<_> = deleted
+            .iter()
+            .chain(&wrapped)
+            .map(|(id, _, _)| *id)
             .collect();
-        let mut deleted_ids: Vec<_> = deleted.keys().copied().collect();
         // update storage
         Arc::get_mut(&mut self.storage)
             .unwrap()
-            .finish(written, deleted);
+            .finish(inner.written, inner.deleted);
         // enumerate objects after written to storage, sort by a "stable" sorting as the
         // object ID is not stable
-        let mut created_ids_vec = created
-            .iter()
-            .map(|((id, _, _), _)| *id)
-            .collect::<Vec<_>>();
+        let mut created_ids_vec = created_set.iter().collect::<Vec<_>>();
         created_ids_vec.sort_by_key(|id| self.get_object_sorting_key(id));
-        for id in &created_ids_vec {
+        for id in created_ids_vec {
             self.enumerate_fake(*id);
         }
         // sort by fake id
diff --git a/crates/sui-types/src/messages.rs b/crates/sui-types/src/messages.rs
index 9907d2ec5dfd1..f4e6f9c9fb99f 100644
--- a/crates/sui-types/src/messages.rs
+++ b/crates/sui-types/src/messages.rs
@@ -12,7 +12,7 @@ use crate::crypto::{
 use crate::gas::GasCostSummary;
 use crate::messages_checkpoint::{CheckpointFragment, CheckpointSequenceNumber};
 use crate::object::{Object, ObjectFormatOptions, Owner, OBJECT_START_VERSION};
-use crate::storage::DeleteKind;
+use crate::storage::{DeleteKind, WriteKind};
 use crate::sui_serde::Base64;
 use crate::SUI_SYSTEM_STATE_OBJECT_ID;
 use base64ct::Encoding;
@@ -1430,11 +1430,16 @@ impl TransactionEffects {
     /// created and unwrapped objects. In other words, all objects that still exist
     /// in the object state after this transaction.
     /// It doesn't include deleted/wrapped objects.
-    pub fn all_mutated(&self) -> impl Iterator<Item = &(ObjectRef, Owner)> + Clone {
+    pub fn all_mutated(&self) -> impl Iterator<Item = (&ObjectRef, &Owner, WriteKind)> + Clone {
         self.mutated
             .iter()
-            .chain(self.created.iter())
-            .chain(self.unwrapped.iter())
+            .map(|(r, o)| (r, o, WriteKind::Mutate))
+            .chain(self.created.iter().map(|(r, o)| (r, o, WriteKind::Create)))
+            .chain(
+                self.unwrapped
+                    .iter()
+                    .map(|(r, o)| (r, o, WriteKind::Unwrap)),
+            )
     }
 
     /// Return an iterator of mutated objects, but excluding the gas object.
@@ -1448,7 +1453,7 @@ impl TransactionEffects {
 
     pub fn is_object_mutated_here(&self, obj_ref: ObjectRef) -> bool {
         // The mutated or created case
-        if self.all_mutated().any(|(oref, _)| *oref == obj_ref) {
+        if self.all_mutated().any(|(oref, _, _)| *oref == obj_ref) {
             return true;
         }
 
diff --git a/crates/sui-types/src/storage.rs b/crates/sui-types/src/storage.rs
index 0b1ade54e9743..a98b63cb02818 100644
--- a/crates/sui-types/src/storage.rs
+++ b/crates/sui-types/src/storage.rs
@@ -8,7 +8,17 @@ use crate::{
     object::Object,
 };
 use serde::{Deserialize, Serialize};
-use std::collections::{BTreeMap, BTreeSet};
+use std::collections::BTreeMap;
+
+#[derive(Debug, PartialEq, Eq, Clone, Copy, Serialize, Deserialize)]
+pub enum WriteKind {
+    /// The object was in storage already but has been modified
+    Mutate,
+    /// The object was created in this transaction
+    Create,
+    /// The object was previously wrapped in another object, but has been restored to storage
+    Unwrap,
+}
 
 #[derive(Debug, PartialEq, Eq, Clone, Copy, Serialize, Deserialize)]
 pub enum DeleteKind {
@@ -22,7 +32,7 @@ pub enum DeleteKind {
 }
 
 pub enum ObjectChange {
-    Write(Object),
+    Write(Object, WriteKind),
     Delete(SequenceNumber, DeleteKind),
 }
 
@@ -32,10 +42,6 @@ pub trait Storage {
 
     fn read_object(&self, id: &ObjectID) -> Option<&Object>;
 
-    // Specify the list of object IDs created during the transaction.
-    // This is needed to determine unwrapped objects at the end.
-    fn set_create_object_ids(&mut self, ids: BTreeSet<ObjectID>);
-
     /// Record an event that happened during execution
     fn log_event(&mut self, event: Event);