fix: Preserve workflow task failure errors through illegal call validator rescues

temporalio/lib/temporalio/internal/worker/workflow_instance/illegal_call_tracer.rb

@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 
+require 'temporalio/internal/workflow_task_failure_error'
 require 'temporalio/worker/illegal_workflow_call_validator'
 require 'temporalio/workflow'
 
@@ -92,6 +93,11 @@ def initialize(illegal_calls)
                                     ))
                     nil
                   rescue Exception => e # rubocop:disable Lint/RescueException
+                    # Errors that invalidate the whole workflow task (e.g. an async DeadlockError landing while the
+                    # validator runs) must propagate with their class intact so they fail the task instead of being
+                    # flattened into a NondeterminismError message that futures may store
+                    raise if e.is_a?(Internal::WorkflowTaskFailureError)
+
                     ", reason: #{e}"
                   end
                 else
@@ -108,6 +114,9 @@ def initialize(illegal_calls)
                                             ))
                       nil
                     rescue Exception => e # rubocop:disable Lint/RescueException
+                      # See workflow-task-failure comment on the class-level validator rescue above
+                      raise if e.is_a?(Internal::WorkflowTaskFailureError)
+
                       ", reason: #{e}"
                     end
                   end

temporalio/test/deadlock_test.rb

@@ -65,6 +65,22 @@ def execute
     end
   end
 
+  module ValidatorInterruptProbe
+    def self.probe; end
+  end
+
+  class DeadlockDuringValidatorWorkflow < Temporalio::Workflow::Definition
+    def execute
+      Temporalio::Workflow::Future.new do
+        ValidatorInterruptProbe.probe
+        Temporalio::Workflow.execute_activity(BasicActivity, 1, start_to_close_timeout: 10)
+      end
+      # Future intentionally not waited (mirrors a fanout blocked on all_of with other
+      # futures pending) so a stored failure would never surface
+      Temporalio::Workflow.wait_condition { false }
+    end
+  end
+
   class ProtobufObjectCachePartialCommandsWorkflow < Temporalio::Workflow::Definition
     ACTIVITY_COUNT = 6
     BLOCK_AFTER_ACTIVITY_COUNT = 3
@@ -144,6 +160,48 @@ def test_deadlock_in_future_fails_workflow_task_and_replays_on_new_worker
     handle&.terminate
   end
 
+  def test_deadlock_during_illegal_call_validator_fails_workflow_task
+    task_queue = "tq-#{SecureRandom.uuid}"
+    # Simulates the deadlock watchdog's async raise landing while an illegal call
+    # validator block is executing (e.g. Thread::Mutex#synchronize inside protobuf's
+    # ObjectCache during ScheduleActivity arg serialization)
+    illegal_calls = Temporalio::Worker.default_illegal_workflow_calls.merge(
+      'DeadlockTest::ValidatorInterruptProbe' => [
+        Temporalio::Worker::IllegalWorkflowCallValidator.new(method_name: :probe) do |_info|
+          raise Temporalio::Worker::WorkflowExecutor::ThreadPool::DeadlockError,
+                '[TMPRL1101] Potential deadlock detected: simulated async deadlock raise'
+        end
+      ]
+    )
+    worker = Temporalio::Worker.new(
+      client: env.client,
+      task_queue:,
+      workflows: [DeadlockDuringValidatorWorkflow],
+      activities: [BasicActivity],
+      illegal_workflow_calls: illegal_calls
+    )
+    # @type var handle: untyped
+    handle = nil
+    worker.run do
+      handle = env.client.start_workflow(
+        DeadlockDuringValidatorWorkflow,
+        id: "wf-#{SecureRandom.uuid}",
+        task_queue:
+      )
+      assert_eventually_task_fail(handle:, message_contains: 'Potential deadlock detected')
+      # The DeadlockError class must survive the tracer. A laundered
+      # NondeterminismError would read "Cannot access ..." and, when raised inside a
+      # future, be stored instead of failing the task at all
+      failure_messages = handle.fetch_history_events.filter_map do |event|
+        event.workflow_task_failed_event_attributes&.failure&.message
+      end
+      refute_empty(failure_messages)
+      failure_messages.each { |message| refute_includes(message, 'Cannot access') }
+    end
+  ensure
+    handle&.terminate
+  end
+
   def test_protobuf_object_cache_mutex_does_not_emit_partial_command_batch
     task_queue = "tq-#{SecureRandom.uuid}"
     # @type var release_mutex: (^() -> void)?