Guard calling after_transition when exception encountered

src/prefect/orion/orchestration/rules.py

@@ -884,12 +884,12 @@ async def __aexit__(
         """
         Exit the async runtime context governed by this transform.
 
-        If the transition has been nullified upon exiting this transforms's context,
+        If the transition has been nullified or errorred upon exiting this transforms's context,
         nothing happens. Otherwise, `self.after_transition` will fire on every non-null
         proposed state.
         """
 
-        if not self.nullified_transition():
+        if not self.nullified_transition() and not self.errorred_transition():
             await self.after_transition(self.context)
             self.context.finalization_signature.append(str(self.__class__))
 
@@ -927,3 +927,13 @@ def nullified_transition(self) -> bool:
         """
 
         return self.context.proposed_state is None
+
+    def errorred_transition(self) -> bool:
+        """
+        Determines if the transition has encountered an exception.
+
+        Returns:
+            True if the transition is encountered an exception, False otherwise.
+        """
+
+        return self.context.orchestration_error is not None

tests/orion/orchestration/test_rules.py

@@ -1086,6 +1086,57 @@ async def after_transition(self, context):
         assert before_hook.call_count == 0
         assert after_hook.call_count == 0
 
+    @pytest.mark.parametrize(
+        "intended_transition",
+        list(product([*states.StateType, None], [*states.StateType])),
+        ids=transition_names,
+    )
+    async def test_universal_transforms_never_fire_after_transition_on_errored_transitions(
+        self, session, task_run, intended_transition
+    ):
+        # nullified transitions occur when the proposed state becomes None
+        # and nothing is written to the database
+
+        side_effect = 0
+        before_hook = MagicMock()
+        after_hook = MagicMock()
+
+        class IllustrativeUniversalTransform(BaseUniversalTransform):
+            async def before_transition(self, context):
+                nonlocal side_effect
+                side_effect += 1
+                before_hook()
+
+            async def after_transition(self, context):
+                nonlocal side_effect
+                side_effect += 1
+                after_hook()
+
+        initial_state_type, proposed_state_type = intended_transition
+        initial_state = await commit_task_run_state(
+            session, task_run, initial_state_type
+        )
+        proposed_state = (
+            states.State(type=proposed_state_type) if proposed_state_type else None
+        )
+
+        ctx = OrchestrationContext(
+            session=session,
+            initial_state=initial_state,
+            proposed_state=proposed_state,
+        )
+
+        universal_transform = IllustrativeUniversalTransform(ctx)
+
+        async with universal_transform as ctx:
+            ctx.orchestration_error = Exception
+
+        assert side_effect == 1
+        assert before_hook.call_count == 1
+        assert (
+            after_hook.call_count == 0
+        ), "after_transition should not be called if orchestration encountered errors."
+
 
 @pytest.mark.parametrize("run_type", ["task", "flow"])
 class TestOrchestrationContext: