@@ -50,7 +50,7 @@
 from airflow.serialization.definitions.dag import SerializedDAG
 from airflow.serialization.definitions.notset import NOTSET, ArgNotSet, is_arg_set
 from airflow.serialization.serialized_objects import DagSerialization, LazyDeserializedDAG
-from airflow.utils.file import correct_maybe_zipped, find_enclosing_file
+from airflow.utils.file import find_enclosing_file
 from airflow.utils.log.logging_mixin import LoggingMixin
 from airflow.utils.session import NEW_SESSION, provide_session
 
@@ -273,9 +273,7 @@ def get_dag(self, dag_id, *, session: Session = NEW_SESSION):
             self.dags.pop(dag_id, None)
         if dag is None or is_expired:
             # Reprocess source file.
-            found_dags = self.process_file(
-                filepath=correct_maybe_zipped(orm_dag.fileloc), only_if_updated=False
-            )
+            found_dags = self.process_file(filepath=orm_dag.fileloc, only_if_updated=False)
 
             # If the source file no longer exports `dag_id`, delete it from self.dags
             if found_dags and dag_id in [found_dag.dag_id for found_dag in found_dags]:
@@ -284,8 +282,8 @@ def get_dag(self, dag_id, *, session: Session = NEW_SESSION):
         return self.dags.get(dag_id)
 
     def process_file(self, filepath, only_if_updated=True, safe_mode=True):
-        """Process a DAG file and return found DAGs."""
-        if filepath is None or not os.path.isfile(filepath):
+        """Process a Dag file, or a Dag definition nested in one, and return found Dags."""
+        if filepath is None or os.path.isdir(filepath):
             return []
         return [
             dag
 
@@ -515,11 +515,17 @@ def deactivate_stale_dags(
             # Dag file's last_finish_time, the Dag is considered stale as has apparently been removed from the file,
             # This is especially relevant for Dag files that generate Dags in a dynamic manner.
             rel_path = Path(dag.relative_fileloc)
-            file_info = DagFileInfo(rel_path=rel_path, bundle_name=dag.bundle_name)
-            if file_info not in last_parsed:
-                # Zip-packaged dags are keyed by the archive path, not the inner file, so try the parent as well
-                file_info = DagFileInfo(rel_path=rel_path.parent, bundle_name=dag.bundle_name)
-            if last_finish_time := last_parsed.get(file_info, None):
+            # A Dag nested in a container (``archive.zip/sub/dag.py``) is parsed under the container.
+            last_finish_time = next(
+                (
+                    last_parsed[file_info]
+                    for candidate in (rel_path, *rel_path.parents)
+                    if (file_info := DagFileInfo(rel_path=candidate, bundle_name=dag.bundle_name))
+                    in last_parsed
+                ),
+                None,
+            )
+            if last_finish_time:
                 if dag.last_parsed_time + timedelta(seconds=self.stale_dag_threshold) < last_finish_time:
                     self.log.info(
                         "Deactivating stale DAG %s. Not parsed for %s seconds (last parsed: %s).",
 
@@ -1137,6 +1137,13 @@ def basic_cycle():
         self.validate_dags(test_dag, found_dags, dagbag, should_be_found=False)
         assert file_path in dagbag.import_errors
 
+    def test_process_file_nested_definition(self, tmp_path, test_zip_path):
+        dagbag = DagBag(dag_folder=os.fspath(tmp_path), collect_dags=False)
+
+        found_dags = dagbag.process_file(os.path.join(test_zip_path, "test_zip.py"))
+
+        assert sorted(dag.dag_id for dag in found_dags) == ["test_zip_autoregister", "test_zip_dag"]
+
     def test_process_file_with_none(self, tmp_path):
         """
         test that process_file can handle Nones
 
@@ -1389,6 +1389,29 @@ def test_deactivate_stale_dags_marks_dags_in_inactive_bundles(self, session):
         )
         assert is_stale_by_dag == {"dag_in_inactive_bundle": True, "dag_in_active_bundle": False}
 
+    @pytest.mark.usefixtures("testing_dag_bundle")
+    def test_deactivate_stale_dags_matches_dags_nested_in_a_container(self, session):
+        session.add(
+            DagModel(
+                dag_id="nested_dag",
+                bundle_name="testing",
+                relative_fileloc="dags.zip/sub/nested_dag.py",
+                last_parsed_time=timezone.utcnow(),
+                is_stale=False,
+            )
+        )
+        session.flush()
+
+        manager = DagFileProcessorManager(max_runs=1, processor_timeout=10 * 60)
+        manager.deactivate_stale_dags(
+            last_parsed={
+                DagFileInfo(rel_path=Path("dags.zip"), bundle_name="testing"): timezone.utcnow()
+                + timedelta(hours=1)
+            }
+        )
+
+        assert session.scalar(select(DagModel.is_stale).where(DagModel.dag_id == "nested_dag"))
+
     @pytest.mark.usefixtures("testing_dag_bundle")
     def test_deactivate_stale_dags_marks_dags_with_null_bundle_name(self, session):
         """Dags carried over from Airflow 2.x keep a NULL bundle_name and must still be deactivated.
@@ -4192,8 +4215,8 @@ def test_refresh_dag_bundles_discovery_failure_keeps_known_files_and_dags(self):
             ),
             mock.patch.object(manager, "update_bundle_state"),
             mock.patch.object(manager, "_find_files_in_bundle", side_effect=OSError("listing failed")),
-            mock.patch.object(manager, "deactivate_deleted_dags") as mock_deactivate,
-            mock.patch.object(manager, "clear_orphaned_import_errors") as mock_clear,
+            mock.patch.object(manager, "deactivate_deleted_dags", autospec=True) as mock_deactivate,
+            mock.patch.object(manager, "clear_orphaned_import_errors", autospec=True) as mock_clear,
             mock.patch.object(manager, "handle_removed_files"),
             mock.patch.object(manager, "_resort_file_queue"),
             mock.patch.object(manager, "_add_new_files_to_queue"),