diff --git a/geonode/api/tests.py b/geonode/api/tests.py
index ec846d6c0db..71559cbe835 100644
--- a/geonode/api/tests.py
+++ b/geonode/api/tests.py
@@ -219,7 +219,7 @@ def test_dataset_get_detail_unauth_dataset_not_public(self):
 
                 self.client.login(username=self.user, password=self.passwd)
                 resp = self.client.get(list_url)
-                self.assertEqual(len(self.deserialize(resp)["objects"]), 7)  # admin can't see resources in dirty_state
+                self.assertEqual(len(self.deserialize(resp)["objects"]), 8)  # admin can see resources in dirty_state
 
                 self.client.logout()
                 resp = self.client.get(list_url)
diff --git a/geonode/layers/api/permissions.py b/geonode/layers/api/permissions.py
index 8ace5857afe..7810e675581 100644
--- a/geonode/layers/api/permissions.py
+++ b/geonode/layers/api/permissions.py
@@ -49,9 +49,9 @@ def filter_queryset(self, request, queryset, view):
             polymorphic_ctype__model="dataset"
         )
         try:
-            include_dirty = strtobool(request.query_params.get("include_dirty", "False"))
+            include_dirty = strtobool(request.query_params.get("include_dirty", "True"))
         except Exception:
-            include_dirty = False
+            include_dirty = True
 
         obj_with_perms = get_visible_resources(
             resources,
diff --git a/geonode/layers/api/tests.py b/geonode/layers/api/tests.py
index 35ded0ef724..934700e402c 100644
--- a/geonode/layers/api/tests.py
+++ b/geonode/layers/api/tests.py
@@ -126,7 +126,7 @@ def test_datasets(self):
     @override_settings(REST_API_DEFAULT_PAGE_SIZE=100)
     def test_filter_dirty_state(self):
         """
-        ensure that a dirty_state dataset wont be returned
+        ensure that a dirty_state dataset will be returned
         """
 
         # ensure that there is atleast one resource with dirty_state
@@ -140,12 +140,11 @@ def test_filter_dirty_state(self):
         self.assertEqual(response.status_code, 200)
         dataset_list = response.data["datasets"]
 
-        # ensure that list count is equal to that of clean data
-        # clean resources
-        resource_count_clean = Dataset.objects.filter(dirty_state=False).count()
+        # dirty resource is now included in count
+        resource_count_clean = Dataset.objects.count()
         self.assertEqual(len(dataset_list), resource_count_clean)
-        # ensure that the updated dirty dataset is not in the response
-        self.assertFalse(dirty_dataset.pk in [int(dataset["pk"]) for dataset in dataset_list])
+        # ensure that the updated dirty dataset is included in the response
+        self.assertTrue(dirty_dataset.pk in [int(dataset["pk"]) for dataset in dataset_list])
 
     @override_settings(REST_API_DEFAULT_PAGE_SIZE=100)
     def test_filter_dirty_state_include_dirty(self):
diff --git a/geonode/security/registry.py b/geonode/security/registry.py
index 4c89e36db60..b2a9705dd91 100644
--- a/geonode/security/registry.py
+++ b/geonode/security/registry.py
@@ -261,7 +261,7 @@ def get_visible_resources(
         admin_approval_required=False,
         unpublished_not_visible=False,
         private_groups_not_visibile=False,
-        include_dirty=False,
+        include_dirty=True,
     ):
         # Get the list of objects the user has access to
         from geonode.groups.models import GroupProfile
diff --git a/geonode/security/utils.py b/geonode/security/utils.py
index 820eef1b00f..8617ee4e92f 100644
--- a/geonode/security/utils.py
+++ b/geonode/security/utils.py
@@ -56,7 +56,7 @@ def get_visible_resources(
     admin_approval_required=False,
     unpublished_not_visible=False,
     private_groups_not_visibile=False,
-    include_dirty=False,
+    include_dirty=True,
 ):
     """
     We should use the registry to get the visible resources.
diff --git a/geonode/upload/celery_tasks.py b/geonode/upload/celery_tasks.py
index ebd82a91b6f..aa172b4aa6a 100644
--- a/geonode/upload/celery_tasks.py
+++ b/geonode/upload/celery_tasks.py
@@ -602,6 +602,13 @@ def copy_geonode_resource(self, exec_id, actual_step, layer_name, alternate, han
 
         _exec = orchestrator.get_execution_object(exec_id)
 
+        # Update input_params with original resource's uuid immediately so the API
+        # can find this execution request for original the dirty resource
+        orchestrator.update_execution_request_status(
+            execution_id=str(_exec.exec_id),
+            input_params={**_exec.input_params, **{"uuid": resource.uuid}},
+        )
+
         workspace = resource.alternate.split(":")[0]
 
         data_to_update = {