Restore shadowed routing state; settle worker failures

lib/cli/ui/spinner/spin_group.rb

@@ -439,6 +439,12 @@ def wait(to: $stdout)
           @work_queue.interrupt
           debrief(to: to) if @interrupt_debrief
           stopped? ? false : raise
+        rescue Exception # rubocop:disable Lint/RescueException
+          # A task failure outside StandardError is not ours to debrief, but it
+          # leaves wait mid-render. Stop the group and its workers before the
+          # exception escapes so sibling tasks do not continue unattended.
+          stop
+          raise
         end
 
         #: (String message) -> void

lib/cli/ui/stdout_router.rb

@@ -209,36 +209,35 @@ def run
 
           StdoutRouter.assert_enabled!
 
-          Thread.current[:cliui_current_capture] = self
-
-          prev_frame_inset = Thread.current[:no_cliui_frame_inset]
-          prev_hook = Thread.current[:cliui_output_hook]
-
-          if Thread.current.respond_to?(:report_on_exception)
-            Thread.current.report_on_exception = false
-          end
-
-          self.class.with_stdin_masked do
-            Thread.current[:no_cliui_frame_inset] = !@with_frame_inset
-            Thread.current[:cliui_output_hook] = ->(data, stream) do
-              stream = :stdout if @merged_output
-              case stream
-              when :stdout
-                @out.write(data)
-                @duplicate_output_to.write(data)
-              when :stderr
-                @err.write(data)
-              else raise
+          previous_capture = Thread.current[:cliui_current_capture]
+          begin
+            Thread.current[:cliui_current_capture] = self
+            self.class.with_stdin_masked do
+              previous_frame_inset = Thread.current[:no_cliui_frame_inset]
+              previous_hook = Thread.current[:cliui_output_hook]
+              begin
+                Thread.current[:no_cliui_frame_inset] = !@with_frame_inset
+                Thread.current[:cliui_output_hook] = ->(data, stream) do
+                  stream = :stdout if @merged_output
+                  case stream
+                  when :stdout
+                    @out.write(data)
+                    @duplicate_output_to.write(data)
+                  when :stderr
+                    @err.write(data)
+                  else raise
+                  end
+                  print_captured_output # suppress writing to terminal by default
+                end
+                @block.call
+              ensure
+                Thread.current[:cliui_output_hook] = previous_hook
+                Thread.current[:no_cliui_frame_inset] = previous_frame_inset
               end
-              print_captured_output # suppress writing to terminal by default
             end
-
-            @block.call
+          ensure
+            Thread.current[:cliui_current_capture] = previous_capture
           end
-        ensure
-          Thread.current[:cliui_output_hook] = prev_hook
-          Thread.current[:no_cliui_frame_inset] = prev_frame_inset
-          Thread.current[:cliui_current_capture] = nil
         end
 
         #: -> String
@@ -318,15 +317,17 @@ class << self
         def with_id(on_streams:, &block)
           require 'securerandom'
           id = format('%05d', rand(10**5))
-          Thread.current[:cliui_output_id] = {
-            id: id,
-            streams: on_streams.map do |stream|
-              stream #: as io_like
-            end,
-          }
-          yield(id)
-        ensure
-          Thread.current[:cliui_output_id] = nil
+          streams = on_streams.map do |stream|
+            stream #: as io_like
+          end
+
+          previous_id = Thread.current[:cliui_output_id]
+          begin
+            Thread.current[:cliui_output_id] = { id: id, streams: streams }
+            yield(id)
+          ensure
+            Thread.current[:cliui_output_id] = previous_id
+          end
         end
 
         #: -> Hash[Symbol, (String | io_like)]?

lib/cli/ui/work_queue.rb

@@ -4,6 +4,10 @@
 module CLI
   module UI
     class WorkQueue
+      # Settled into a future whose worker left without settling it itself.
+      class WorkerDied < StandardError
+      end
+
       class Future
         #: -> void
         def initialize
@@ -69,17 +73,19 @@ def initialize(max_concurrent)
         @max_concurrent = max_concurrent
         @queue = Queue.new #: Queue
         @mutex = Mutex.new #: Mutex
+        @interrupt_mutex = Mutex.new #: Mutex
         @condition = ConditionVariable.new #: ConditionVariable
         @workers = [] #: Array[Thread]
+        @stopping = false #: bool
       end
 
       #: { -> untyped } -> Future
       def enqueue(&block)
         future = Future.new
         @mutex.synchronize do
+          @queue.push([future, block])
           start_worker if @workers.size < @max_concurrent
         end
-        @queue.push([future, block])
         future
       end
 
@@ -91,22 +97,49 @@ def close
       #: -> void
       def wait
         @queue.close
-        @workers.each(&:join)
+        loop do
+          workers = @mutex.synchronize { @workers.dup }
+          break if workers.empty?
+
+          workers.each(&:join)
+        end
       end
 
       #: -> void
       def interrupt
-        @mutex.synchronize do
-          @queue.close
-          # Fail any remaining tasks in the queue
-          until @queue.empty?
-            future, _block = @queue.pop(true)
-            future&.fail(Interrupt.new)
+        @interrupt_mutex.synchronize do
+          workers = @mutex.synchronize do
+            @stopping = true
+            @queue.close
+
+            # Fail any remaining tasks in the queue. Workers can consume from
+            # the queue concurrently, so an empty? check followed by pop is racy.
+            loop do
+              future, _block = @queue.pop(true)
+              future&.fail(Interrupt.new)
+            rescue ThreadError
+              break
+            end
+
+            @workers.dup
           end
-          # Interrupt all worker threads
-          @workers.each { |worker| worker.raise(Interrupt) if worker.alive? }
-          @workers.each(&:join)
-          @workers.clear
+
+          # These are WorkQueue-owned threads being deliberately torn down, so
+          # neither their thread-death report nor their Interrupt belongs to the
+          # caller performing the teardown.
+          workers.each do |worker|
+            next unless worker.alive?
+
+            worker.report_on_exception = false
+            worker.raise(Interrupt)
+          end
+          workers.each do |worker|
+            worker.join
+          rescue Interrupt
+            nil
+          end
+        ensure
+          @mutex.synchronize { @workers.clear }
         end
       end
 
@@ -115,26 +148,59 @@ def interrupt
       #: -> void
       def start_worker
         @workers << Thread.new do
-          loop do
-            work = @queue.pop
-            break if work.nil?
-
-            future, block = work
+          run_worker
+        rescue Interrupt
+          # Clean exit on interrupt
+        ensure
+          worker_finished(Thread.current)
+        end
+      end
 
-            begin
+      #: -> void
+      def run_worker
+        loop do
+          future = nil #: Future?
+          begin
+            # Do not let an asynchronous exception land after Queue#pop has
+            # removed work but before its future is assigned. Interrupts remain
+            # enabled while pop is blocked and while the task itself is running.
+            Thread.handle_interrupt(Exception => :never) do
+              work = Thread.handle_interrupt(Exception => :on_blocking) { @queue.pop }
+              return if work.nil?
+
+              future, block = work
               future.start
-              result = block.call
+              result = Thread.handle_interrupt(Exception => :immediate) { block.call }
               future.complete(result)
-            rescue Interrupt => e
-              future.fail(e)
-              raise # Always re-raise interrupts to terminate the worker
-            rescue StandardError => e
-              future.fail(e)
-              # Don't re-raise standard errors - allow worker to continue
+            end
+          rescue Interrupt => e
+            future&.fail(e)
+            raise # Always re-raise interrupts to terminate the worker
+          rescue Exception => e # rubocop:disable Lint/RescueException
+            # The future carries the error to callers. Keep the worker: tasks
+            # already queued may have no later enqueue to replace it.
+            future&.fail(e)
+          ensure
+            if future
+              Thread.handle_interrupt(Exception => :never) do
+                unless future.completed?
+                  future.fail(WorkerDied.new('worker died before its task completed'))
+                end
+              end
+            end
+          end
+        end
+      end
+
+      #: (Thread worker) -> void
+      def worker_finished(worker)
+        Thread.handle_interrupt(Exception => :never) do
+          @mutex.synchronize do
+            @workers.delete(worker)
+            if !@stopping && !@queue.empty? && @workers.size < @max_concurrent
+              start_worker
             end
           end
-        rescue Interrupt
-          # Clean exit on interrupt
         end
       end
     end

test/cli/ui/spinner/spin_group_test.rb

@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 require 'test_helper'
+require 'timeout'
 
 module CLI
   module UI
@@ -36,6 +37,45 @@ def test_spin_group_auto_debrief_false
           assert_equal('', err)
         end
 
+        def test_spin_group_non_standard_error_does_not_hang_or_report_thread_death
+          _out, err = capture_io do
+            CLI::UI::StdoutRouter.ensure_activated
+
+            sg = SpinGroup.new(auto_debrief: false)
+            sg.add('s') { raise NotImplementedError, 'not implemented' }
+
+            error = Timeout.timeout(10) do
+              assert_raises(NotImplementedError) { sg.wait }
+            end
+            assert_equal('not implemented', error.message)
+          end
+
+          assert_equal('', err)
+        end
+
+        def test_spin_group_non_standard_error_stops_the_group_and_its_siblings
+          _out, err = capture_io do
+            CLI::UI::StdoutRouter.ensure_activated
+
+            sg = SpinGroup.new(auto_debrief: false)
+            sibling_finished = false
+            sg.add('boom') { raise NotImplementedError, 'not implemented' }
+            sg.add('sibling') do
+              sleep(30)
+              sibling_finished = true
+            end
+
+            Timeout.timeout(10) do
+              assert_raises(NotImplementedError) { sg.wait }
+            end
+
+            assert(sg.stopped?)
+            refute(sibling_finished)
+          end
+
+          assert_equal('', err)
+        end
+
         def test_spin_group_success_debrief
           capture_io do
             CLI::UI::StdoutRouter.ensure_activated