diff --git a/src/python/pants/engine/internals/engine_test.py b/src/python/pants/engine/internals/engine_test.py
index 2c6d396e9d55..3c7a182e0d28 100644
--- a/src/python/pants/engine/internals/engine_test.py
+++ b/src/python/pants/engine/internals/engine_test.py
@@ -6,7 +6,7 @@
 import unittest
 from dataclasses import dataclass, field
 from textwrap import dedent
-from typing import List, Optional
+from typing import List, Optional, Tuple
 
 from pants.engine.engine_aware import EngineAwareReturnType
 from pants.engine.fs import (
@@ -16,6 +16,7 @@
     Digest,
     DigestContents,
     FileContent,
+    MergeDigests,
     Snapshot,
 )
 from pants.engine.internals.engine_testutil import (
@@ -167,6 +168,30 @@ def rule_C(e: Epsilon) -> Alpha:
     return Alpha()
 
 
+@dataclass(frozen=True)
+class FileInput:
+    filename: str
+
+
+@dataclass(frozen=True)
+class MergedOutput:
+    digest: Digest
+
+
+@rule
+async def catch_merge_digests_error(file_input: FileInput) -> MergedOutput:
+    # Create two separate digests writing different contents to the same file path.
+    input_1 = CreateDigest((FileContent(path=file_input.filename, content=b"yes"),))
+    input_2 = CreateDigest((FileContent(path=file_input.filename, content=b"no"),))
+    digests = await MultiGet(Get(Digest, CreateDigest, input_1),
+                             Get(Digest, CreateDigest, input_2))
+    try:
+        merged = await Get(Digest, MergeDigests(digests))
+    except Exception as e:
+        raise Exception(f'error merging digests for input {file_input}: {e}')
+    return MergedOutput(merged)
+
+
 class EngineTest(unittest.TestCase, SchedulerTestBase):
 
     assert_equal_with_printing = assert_equal_with_printing
@@ -250,6 +275,13 @@ def test_missing_query_rule(self) -> None:
             "for MyInt)."
         ) in str(cm.exception)
 
+    def test_catch_intrinsic_error(self) -> None:
+        rules = [catch_merge_digests_error, QueryRule(MergedOutput, (FileInput,))]
+        scheduler = self.mk_scheduler(rules=rules, include_trace_on_error=False)
+        with self.assertRaises(Exception) as cm:
+            scheduler.product_request(MergedOutput, subjects=[FileInput('some-file.txt')])
+        assert "error merging digests for input FileInput(filename=\'some-file.txt\')" in str(cm.exception)
+
 
 @dataclass
 class WorkunitTracker: