diff --git a/CHANGELOG.md b/CHANGELOG.md
index 60e37a5ebf32..fbf33ca9b6d6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,13 @@
 
 ## Next Release
 
+### Fixes to Crashes
+
+- Report a clean error instead of an internal error when the mypy cache is
+  corrupt (for example, if the cache directory was partially overwritten).
+  Mypy now tells you to remove the cache directory (or pass a different
+  `--cache-dir`) and re-run, instead of crashing with `INTERNAL ERROR`.
+
 ## Mypy 2.4
 
 We've just uploaded mypy 2.4.0 to the Python Package Index ([PyPI](https://pypi.org/project/mypy/)).
diff --git a/mypy/build.py b/mypy/build.py
index 78bd732793c2..b1b5c7b0d391 100644
--- a/mypy/build.py
+++ b/mypy/build.py
@@ -1821,6 +1821,31 @@ def _load_ff_file(
     return data
 
 
+def _report_corrupt_cache_error(manager: BuildManager, file: str) -> None:
+    """Report a blocking error for a cache file that can't be read or deserialized."""
+    manager.errors.set_file(file, None, manager.options)
+    manager.error(
+        None,
+        "Error reading cache file;"
+        " you likely have a corrupt cache.\n"
+        "Try removing the {cache_dir} directory"
+        " and run mypy again.".format(cache_dir=manager.options.cache_dir),
+        blocker=True,
+    )
+
+
+def _is_corrupt_cache_assertion(err: BaseException) -> bool:
+    """Return True if the exception indicates a partially deserialized (corrupt) cache.
+
+    During deserialization mypy leaves ``Instance`` types pointing at a ``NOT_READY``
+    placeholder and only resolves them in a later fix-up pass. If that pass fails to
+    find the referenced type, any later attribute access on the placeholder raises an
+    ``AssertionError`` with this exact message. That only happens for cached modules,
+    so it is a reliable signal that the cache is corrupt.
+    """
+    return isinstance(err, AssertionError) and str(err).startswith("De-serialization failure")
+
+
 def _load_json_file(
     file: str, manager: BuildManager, log_success: str, log_error: str
 ) -> dict[str, Any] | None:
@@ -1840,15 +1865,7 @@ def _load_json_file(
         result = json_loads(data)
         manager.add_stats(data_file_load_time=time.time() - t1)
     except json.JSONDecodeError:
-        manager.errors.set_file(file, None, manager.options)
-        manager.error(
-            None,
-            "Error reading JSON file;"
-            " you likely have a bad cache.\n"
-            "Try removing the {cache_dir} directory"
-            " and run mypy again.".format(cache_dir=manager.options.cache_dir),
-            blocker=True,
-        )
+        _report_corrupt_cache_error(manager, file)
         return None
     else:
         assert isinstance(result, dict)
@@ -2026,9 +2043,17 @@ def find_cache_meta(
             manager.log(f"Metadata abandoned for {id}: incompatible cache format")
             return None
         data_io = ReadBuffer(meta[2:])
-        m = CacheMeta.read(data_io, data_file)
+        try:
+            m = CacheMeta.read(data_io, data_file)
+        except Exception:
+            _report_corrupt_cache_error(manager, meta_file)
+            return None
     else:
-        m = CacheMeta.deserialize(meta, data_file)
+        try:
+            m = CacheMeta.deserialize(meta, data_file)
+        except Exception:
+            _report_corrupt_cache_error(manager, meta_file)
+            return None
     if m is None:
         manager.log(f"Metadata abandoned for {id}: cannot deserialize data")
         return None
@@ -2118,9 +2143,17 @@ def find_cache_meta(
         return None
     if isinstance(meta_ex, bytes):
         data_io = ReadBuffer(meta_ex)
-        me = CacheMetaEx.read(data_io)
+        try:
+            me = CacheMetaEx.read(data_io)
+        except Exception:
+            _report_corrupt_cache_error(manager, meta_ex_file)
+            return None
     else:
-        me = CacheMetaEx.deserialize(meta_ex)
+        try:
+            me = CacheMetaEx.deserialize(meta_ex)
+        except Exception:
+            _report_corrupt_cache_error(manager, meta_ex_file)
+            return None
     if me is None:
         return None
     manager.add_stats(fresh_metas=1)
@@ -3095,6 +3128,13 @@ def wrap_context(self, check_blockers: bool = True) -> Iterator[None]:
         except CompileError:
             raise
         except Exception as err:
+            if _is_corrupt_cache_assertion(err):
+                # A cached module was only partially deserialized (likely a corrupt
+                # cache), so it leaks a ``NOT_READY`` placeholder that blows up on the
+                # first attribute access. Report a clean, actionable error instead of
+                # an internal error.
+                _report_corrupt_cache_error(self.manager, self.path or "<unknown>")
+                self.check_blockers()
             report_internal_error(
                 err,
                 self.path,
@@ -3132,11 +3172,19 @@ def load_tree(self, temporary: bool = False) -> None:
 
         t0 = time.time()
         # TODO: Assert data file wasn't changed.
-        if isinstance(data, bytes):
-            data_io = ReadBuffer(data)
-            self.tree = MypyFile.read(data_io)
-        else:
-            self.tree = MypyFile.deserialize(data)
+        try:
+            if isinstance(data, bytes):
+                data_io = ReadBuffer(data)
+                self.tree = MypyFile.read(data_io)
+            else:
+                self.tree = MypyFile.deserialize(data)
+        except Exception:
+            # If the cache data file can't be deserialized, it is likely corrupt.
+            # Report a clean error instead of failing with an internal error.
+            _report_corrupt_cache_error(self.manager, data_file)
+            # Abort the build right away, as the module tree is unusable.
+            self.check_blockers()
+            raise
         t1 = time.time()
         self.manager.add_stats(deserialize_time=t1 - t0)
         if not temporary:
@@ -3147,8 +3195,25 @@ def fix_cross_refs(self) -> None:
         assert self.tree is not None, "Internal error: method must be called on parsed file only"
         # Do initial lightweight pass fixing TypeInfos and module cross-references.
         assert modules_state.node_fixer is not None
-        modules_state.node_fixer.visit_symbol_table(self.tree.names)
-        type_fixer = modules_state.node_fixer.type_fixer
+        try:
+            self._fix_cross_refs()
+        except Exception:
+            # Failing to resolve cross-references almost always means the cache is
+            # corrupt. Report a clean, actionable error instead of crashing with an
+            # internal error.
+            _report_corrupt_cache_error(
+                self.manager,
+                self.meta.data_file if self.meta is not None else (self.path or "<unknown>"),
+            )
+            self.check_blockers()
+            raise
+
+    def _fix_cross_refs(self) -> None:
+        node_fixer = modules_state.node_fixer
+        assert node_fixer is not None
+        assert self.tree is not None
+        node_fixer.visit_symbol_table(self.tree.names)
+        type_fixer = node_fixer.type_fixer
         # Eagerly fix shared instances, before they are used by named_type() calls.
         if instance_cache.str_type is not None:
             instance_cache.str_type.accept(type_fixer)
diff --git a/mypy/test/testcache.py b/mypy/test/testcache.py
new file mode 100644
index 000000000000..ec3df0345c59
--- /dev/null
+++ b/mypy/test/testcache.py
@@ -0,0 +1,92 @@
+"""Unit tests for mypy cache behavior."""
+
+from __future__ import annotations
+
+import os
+import shutil
+import tempfile
+import unittest
+
+from mypy import build
+from mypy.errors import CompileError
+from mypy.modulefinder import BuildSource
+from mypy.options import Options
+
+DEP_SOURCE = "x: int = 1\n"
+
+
+class CorruptCacheSuite(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tempdir = tempfile.mkdtemp()
+        self.cache_dir = os.path.join(self.tempdir, ".mypy_cache")
+        self.mypy_path = os.path.join(self.tempdir, "src")
+        os.mkdir(self.mypy_path)
+        self.dep_path = os.path.join(self.mypy_path, "dep.py")
+        with open(self.dep_path, "w") as f:
+            f.write(DEP_SOURCE)
+        # Messages flushed during a build, including from failing builds.
+        self.flushed: list[str] = []
+
+    def tearDown(self) -> None:
+        shutil.rmtree(self.tempdir)
+
+    def make_options(self) -> Options:
+        options = Options()
+        options.cache_dir = self.cache_dir
+        options.mypy_path = [self.mypy_path]
+        options.hide_error_codes = True
+        # Use the file-based fixed-format cache so we can corrupt the cache
+        # data files directly.
+        options.sqlite_cache = False
+        options.fixed_format_cache = True
+        return options
+
+    def run_build(self, options: Options, main_source: str) -> bool:
+        """Run a build, collecting flushed messages.
+
+        Return True if the build failed with a (clean) CompileError, False if
+        it succeeded.
+        """
+        sources = [BuildSource("main", "__main__", main_source)]
+        try:
+            build.build(
+                sources=sources,
+                options=options,
+                flush_errors=lambda _fn, msgs, _serious: self.flushed.extend(msgs),
+            )
+        except CompileError:
+            return True
+        return False
+
+    def data_file(self, options: Options) -> str:
+        cache_prefix = os.path.join(self.cache_dir, "%d.%d" % options.python_version)
+        return os.path.join(cache_prefix, "dep.data.ff")
+
+    def corrupt_dep_data_file(self, options: Options) -> None:
+        data_file = self.data_file(options)
+        assert os.path.isfile(data_file)
+        st = os.stat(data_file)
+        with open(data_file, "wb") as f:
+            f.write(b"this is not a valid cache data file")
+        # Keep the mtime intact so that the cache metadata still validates.
+        os.utime(data_file, (st.st_atime, st.st_mtime))
+
+    def test_corrupt_cache_data_file(self) -> None:
+        # Populate the cache.
+        options = self.make_options()
+        assert not self.run_build(options, "import dep\n")
+
+        # Corrupt the cached data of the dep module.
+        self.corrupt_dep_data_file(options)
+
+        # A module that requires the (now corrupt) dep cache to be loaded must
+        # produce a clean, actionable error instead of an INTERNAL ERROR.
+        self.flushed = []
+        failed = self.run_build(options, "import dep\ny: int = dep.x\n")
+        assert failed, "build with corrupt cache should fail"
+        assert any("corrupt cache" in msg for msg in self.flushed), self.flushed
+        assert not any("INTERNAL ERROR" in msg for msg in self.flushed), self.flushed
+
+
+if __name__ == "__main__":
+    unittest.main()