diff --git a/gems/karya/spec/e2e/karya/cli_worker_mysql_spec.rb b/gems/karya/spec/e2e/karya/cli_worker_mysql_spec.rb
index ef96334a..e71cddcb 100644
--- a/gems/karya/spec/e2e/karya/cli_worker_mysql_spec.rb
+++ b/gems/karya/spec/e2e/karya/cli_worker_mysql_spec.rb
@@ -7,6 +7,7 @@
 
 require_relative '../spec_helper'
 require File.expand_path('../../../../../spec/support/mysql_e2e_support', __dir__)
+require File.expand_path('../../../../../spec/support/e2e_subprocess', __dir__)
 
 RSpec.describe Karya::CLI, :e2e, :integration do
   def mysql_boot_file(mysql_url:, namespace:, marker_file:)
@@ -52,7 +53,7 @@ def self.call(account_id:, marker_path:)
 
         File.write(boot_file, mysql_boot_file(mysql_url: database_url, namespace:, marker_file:))
 
-        stdout, stderr, status = Open3.capture3(
+        stdout, stderr, status = KaryaSpecSupport::E2ESubprocess.capture(
           *karya_command(
             'worker',
             'billing',
diff --git a/gems/karya/spec/e2e/karya/cli_worker_redis_spec.rb b/gems/karya/spec/e2e/karya/cli_worker_redis_spec.rb
index bde6fff9..e7aa8545 100644
--- a/gems/karya/spec/e2e/karya/cli_worker_redis_spec.rb
+++ b/gems/karya/spec/e2e/karya/cli_worker_redis_spec.rb
@@ -6,6 +6,7 @@
 # LICENSE file in the root directory of this source tree.
 
 require_relative '../spec_helper'
+require File.expand_path('../../../../../spec/support/e2e_subprocess', __dir__)
 
 RSpec.describe Karya::CLI, :e2e, :integration do
   def delete_redis_namespace(redis_url:, namespace:)
@@ -58,7 +59,7 @@ def self.call(account_id:, marker_path:)
 
       File.write(boot_file, redis_boot_file(redis_url:, namespace:, marker_file:))
 
-      stdout, stderr, status = Open3.capture3(
+      stdout, stderr, status = KaryaSpecSupport::E2ESubprocess.capture(
         *karya_command(
           'worker',
           'billing',
diff --git a/gems/karya/spec/e2e/karya/cli_worker_spec.rb b/gems/karya/spec/e2e/karya/cli_worker_spec.rb
index 5aa3b727..43014ff4 100644
--- a/gems/karya/spec/e2e/karya/cli_worker_spec.rb
+++ b/gems/karya/spec/e2e/karya/cli_worker_spec.rb
@@ -6,10 +6,11 @@
 # LICENSE file in the root directory of this source tree.
 
 require_relative '../spec_helper'
+require File.expand_path('../../../../../spec/support/e2e_subprocess', __dir__)
 
 RSpec.describe Karya::CLI, :e2e, :integration do
   def run_cli(*args)
-    Open3.capture3(*karya_command(*args), chdir: KaryaE2EHelpers::PACKAGE_ROOT)
+    KaryaSpecSupport::E2ESubprocess.capture(*karya_command(*args), chdir: KaryaE2EHelpers::PACKAGE_ROOT)
   end
 
   def wait_for_runtime_phase(state_file, *phases)
@@ -41,7 +42,7 @@ def build_force_stop_boot_file(directory:, marker_file:, draining_marker_file:)
   end
 
   def with_force_stop_worker(boot_file:, state_file:, &)
-    Open3.popen2e(*karya_command(
+    process = KaryaSpecSupport::E2ESubprocess.new(*karya_command(
       'worker',
       'billing',
       '--require',
@@ -58,7 +59,10 @@ def with_force_stop_worker(boot_file:, state_file:, &)
       '0',
       '--state-file',
       state_file
-    ), chdir: KaryaE2EHelpers::PACKAGE_ROOT, &)
+    ), chdir: KaryaE2EHelpers::PACKAGE_ROOT)
+    yield process
+  ensure
+    KaryaSpecSupport::E2ESubprocess.close_preserving_failure(process)
   end
 
   def request_force_stop(supervisor_pid:, draining_marker_file:)
@@ -69,39 +73,27 @@ def request_force_stop(supervisor_pid:, draining_marker_file:)
     nil
   end
 
-  def cleanup_worker_process(wait_thr)
-    return unless wait_thr.alive?
-
-    Process.kill('TERM', wait_thr.pid)
-    sleep(0.1)
-    Process.kill('KILL', wait_thr.pid) if wait_thr.alive?
-  rescue Errno::ESRCH
-    nil
-  end
-
-  def expect_force_stopped_worker(wait_thr:, stdout_and_stderr:, state_file:, marker_file:)
-    process_status = Timeout.timeout(10) { wait_thr.value }
-    output = stdout_and_stderr.read
+  def expect_force_stopped_worker(process:, state_file:, marker_file:)
+    process_status = process.wait(timeout: 10)
+    begin
+      process.wait_for_output
+    rescue Timeout::Error
+      raise Timeout::Error, "force-stopped worker output remained open:\n#{process.output}"
+    end
     runtime_state = wait_for_runtime_phase(state_file, 'stopped', 'force_stopping')
 
-    expect(process_status.exitstatus).to eq(1), -> { "worker output:\n#{output}" }
+    expect(process_status.exitstatus).to eq(1), -> { "worker output:\n#{process.output}" }
     expect(File.read(marker_file).strip).not_to be_empty
     expect(runtime_state.fetch('phase')).to match(/\A(?:stopped|force_stopping)\z/)
   end
 
-  def with_force_stop_cleanup(wait_thr:, stdout_and_stderr:, state_file:, marker_file:)
+  def with_force_stop_cleanup(process:, state_file:, marker_file:)
     yield
     expect_force_stopped_worker(
-      wait_thr:,
-      stdout_and_stderr:,
+      process:,
       state_file:,
       marker_file:
     )
-  rescue Timeout::Error, RSpec::Expectations::ExpectationNotMetError
-    cleanup_worker_process(wait_thr)
-    raise
-  ensure
-    cleanup_worker_process(wait_thr)
   end
 
   it 'executes a queued job end-to-end through exe/karya worker' do
@@ -154,10 +146,9 @@ def with_force_stop_cleanup(wait_thr:, stdout_and_stderr:, state_file:, marker_f
       draining_marker_file = File.join(directory, 'draining.txt')
       boot_file = build_force_stop_boot_file(directory:, marker_file:, draining_marker_file:)
 
-      with_force_stop_worker(boot_file:, state_file:) do |_stdin, stdout_and_stderr, wait_thr|
+      with_force_stop_worker(boot_file:, state_file:) do |process|
         with_force_stop_cleanup(
-          wait_thr:,
-          stdout_and_stderr:,
+          process:,
           state_file:,
           marker_file:
         ) do
@@ -176,7 +167,7 @@ def with_force_stop_cleanup(wait_thr:, stdout_and_stderr:, state_file:, marker_f
       draining_marker_file = File.join(directory, 'draining.txt')
       boot_file = build_force_stop_boot_file(directory:, marker_file:, draining_marker_file:)
 
-      with_force_stop_worker(boot_file:, state_file:) do |_stdin, stdout_and_stderr, wait_thr|
+      with_force_stop_worker(boot_file:, state_file:) do
         expect(wait_until { File.exist?(marker_file) && File.exist?(state_file) }).to be(true)
 
         inspect_stdout, inspect_stderr, inspect_status = run_cli('runtime', 'inspect', '--state-file', state_file)
@@ -195,8 +186,6 @@ def with_force_stop_cleanup(wait_thr:, stdout_and_stderr:, state_file:, marker_f
         expect(force_stop_status.exitstatus).to eq(0), -> { "stdout:\n#{force_stop_stdout}\n\nstderr:\n#{force_stop_stderr}" }
         runtime_phase = wait_for_runtime_phase(state_file, 'force_stopping', 'stopped').fetch('phase')
         expect(runtime_phase).to match(/\A(?:force_stopping|stopped)\z/)
-        cleanup_worker_process(wait_thr)
-        stdout_and_stderr.read
       end
     end
   end
diff --git a/gems/karya/spec/e2e/karya/cli_worker_sqlite_spec.rb b/gems/karya/spec/e2e/karya/cli_worker_sqlite_spec.rb
index 165e057d..cd617a78 100644
--- a/gems/karya/spec/e2e/karya/cli_worker_sqlite_spec.rb
+++ b/gems/karya/spec/e2e/karya/cli_worker_sqlite_spec.rb
@@ -7,6 +7,7 @@
 
 require_relative '../spec_helper'
 require File.expand_path('../../../../../spec/support/sqlite_e2e_support', __dir__)
+require File.expand_path('../../../../../spec/support/e2e_subprocess', __dir__)
 
 RSpec.describe Karya::CLI, :e2e, :integration do
   def sqlite_boot_file(sqlite_url:, namespace:, marker_file:)
@@ -50,7 +51,7 @@ def self.call(account_id:, marker_path:)
 
         File.write(boot_file, sqlite_boot_file(sqlite_url: database_url, namespace:, marker_file:))
 
-        stdout, stderr, status = Open3.capture3(
+        stdout, stderr, status = KaryaSpecSupport::E2ESubprocess.capture(
           *karya_command(
             'worker',
             'billing',
diff --git a/gems/karya/spec/e2e/karya_spec_support/e2_e_subprocess_spec.rb b/gems/karya/spec/e2e/karya_spec_support/e2_e_subprocess_spec.rb
new file mode 100644
index 00000000..1b942214
--- /dev/null
+++ b/gems/karya/spec/e2e/karya_spec_support/e2_e_subprocess_spec.rb
@@ -0,0 +1,66 @@
+# frozen_string_literal: true
+
+# Copyright Codevedas Inc. 2025-present
+#
+# This source code is licensed under the MIT license found in the
+# LICENSE file in the root directory of this source tree.
+
+require_relative '../spec_helper'
+require File.expand_path('../../../../../spec/support/e2e_subprocess', __dir__)
+require 'English'
+
+RSpec.describe KaryaSpecSupport::E2ESubprocess, :e2e, :integration do
+  it 'bounds capture when a forked descendant retains an output pipe' do
+    command = [
+      RbConfig.ruby,
+      '-e',
+      <<~RUBY
+        fork do
+          Signal.trap('TERM', 'IGNORE')
+          loop { sleep 0.1 }
+        end
+      RUBY
+    ]
+
+    expect do
+      Timeout.timeout(5) do
+        described_class.capture(*command, timeout: 0.5)
+      end
+    end.to raise_error(Timeout::Error, /subprocess timed out after 0.5s/)
+  end
+
+  it 'terminates a descendant that outlives its leader without retaining output pipes' do
+    command = [
+      RbConfig.ruby,
+      '-e',
+      <<~RUBY
+        fork do
+          1.upto(255) do |file_descriptor|
+            IO.for_fd(file_descriptor).close
+          rescue Errno::EBADF
+            nil
+          end
+          Signal.trap('TERM', 'IGNORE')
+          loop { sleep 0.1 }
+        end
+        exit! 0
+      RUBY
+    ]
+
+    _stdout, _stderr, status = described_class.capture(*command, timeout: 2)
+
+    expect(status).to be_success
+  end
+
+  it 'preserves an active failure when cleanup also fails' do
+    process = instance_double(described_class)
+    cleanup_error = described_class::CleanupError.new('cleanup failed')
+    allow(process).to receive(:close).and_raise(cleanup_error)
+
+    expect do
+      raise 'primary failure'
+    ensure
+      described_class.close_preserving_failure(process, active_exception: $ERROR_INFO)
+    end.to raise_error(RuntimeError, 'primary failure').and output(/cleanup failed/).to_stderr
+  end
+end
diff --git a/spec/support/e2e_subprocess.rb b/spec/support/e2e_subprocess.rb
new file mode 100644
index 00000000..e6165507
--- /dev/null
+++ b/spec/support/e2e_subprocess.rb
@@ -0,0 +1,195 @@
+# frozen_string_literal: true
+
+# Copyright Codevedas Inc. 2025-present
+#
+# This source code is licensed under the MIT license found in the
+# LICENSE file in the root directory of this source tree.
+
+require 'open3'
+require 'timeout'
+
+module KaryaSpecSupport
+  class E2ESubprocess
+    class CleanupError < StandardError; end
+    class OutputTimeout < Timeout::Error; end
+
+    DEFAULT_TIMEOUT = 30
+    TERMINATION_TIMEOUT = 2
+
+    attr_reader :pid
+
+    def self.capture(*command, timeout: DEFAULT_TIMEOUT, **options)
+      deadline = monotonic_time + timeout
+      cleanup_reserve = [TERMINATION_TIMEOUT, timeout / 2.0].min
+      operation_deadline = deadline - cleanup_reserve
+      process = new(*command, **options)
+      begin
+        status = process.wait(timeout: remaining_time(operation_deadline))
+        process.wait_for_output(timeout: remaining_time(operation_deadline))
+        result = [process.stdout, process.stderr, status]
+      rescue Timeout::Error
+        failure = Timeout::Error.new("subprocess timed out after #{timeout}s:\n#{process.output}")
+      ensure
+        begin
+          process.close(deadline:)
+        rescue CleanupError => error
+          failure = failure ? CleanupError.new("#{failure.message}\n#{error.message}") : error
+        end
+      end
+
+      raise failure if failure
+
+      result
+    end
+
+    def self.close_preserving_failure(process, active_exception: $!)
+      process&.close
+    rescue CleanupError => error
+      raise error unless active_exception
+
+      warn "#{error.class}: #{error.message}"
+    end
+
+    def self.monotonic_time
+      Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    end
+    private_class_method :monotonic_time
+
+    def self.remaining_time(deadline)
+      remaining = deadline - monotonic_time
+      raise Timeout::Error unless remaining.positive?
+
+      remaining
+    end
+    private_class_method :remaining_time
+
+    def initialize(*command, **options)
+      @stdin, @stdout_io, @stderr_io, @wait_thread = Open3.popen3(*command, **options.merge(pgroup: true))
+      @pid = @wait_thread.pid
+      @stdout = +''
+      @stderr = +''
+      @output_lock = Mutex.new
+      @stdout_reader = read_output(@stdout_io, @stdout)
+      @stderr_reader = read_output(@stderr_io, @stderr)
+      @stdin.close
+    end
+
+    def alive?
+      @wait_thread.alive?
+    end
+
+    def wait(timeout: DEFAULT_TIMEOUT)
+      Timeout.timeout(timeout) { @wait_thread.value }
+    end
+
+    def wait_for_output(timeout: TERMINATION_TIMEOUT)
+      Timeout.timeout(timeout, OutputTimeout) do
+        @stdout_reader.value
+        @stderr_reader.value
+      end
+    end
+
+    def stdout
+      output_snapshot(@stdout)
+    end
+
+    def stderr
+      output_snapshot(@stderr)
+    end
+
+    def output
+      [stdout, stderr].reject(&:empty?).join("\n")
+    end
+
+    def close(deadline: monotonic_time + TERMINATION_TIMEOUT)
+      terminate_process_group(deadline) unless shutdown_complete?
+      close_output_streams
+      join_until(@wait_thread, deadline)
+      join_until(@stdout_reader, deadline)
+      join_until(@stderr_reader, deadline)
+      wait_for_shutdown(deadline)
+      verify_closed
+    end
+
+    private
+
+    def read_output(stream, destination)
+      Thread.new do
+        loop do
+          chunk = stream.readpartial(4096)
+          @output_lock.synchronize { destination << chunk }
+        end
+      rescue EOFError, IOError
+        nil
+      end
+    end
+
+    def output_snapshot(buffer)
+      @output_lock.synchronize { buffer.dup }
+    end
+
+    def output_readers_alive?
+      @stdout_reader.alive? || @stderr_reader.alive?
+    end
+
+    def terminate_process_group(deadline)
+      signal_process_group('TERM')
+      grace_deadline = monotonic_time + (remaining_time(deadline) / 2.0)
+      join_until(@wait_thread, grace_deadline)
+      return if shutdown_complete?
+
+      signal_process_group('KILL')
+    end
+
+    def signal_process_group(signal)
+      Process.kill(signal, -pid)
+    rescue Errno::ESRCH
+      nil
+    end
+
+    def close_output_streams
+      @stdout_io.close unless @stdout_io.closed?
+      @stderr_io.close unless @stderr_io.closed?
+    end
+
+    def join_until(thread, deadline)
+      thread.join([remaining_time(deadline), 0].max)
+    end
+
+    def verify_closed
+      return if shutdown_complete?
+
+      raise CleanupError, "subprocess cleanup incomplete for process group #{pid}:\n#{output}"
+    end
+
+    def wait_for_shutdown(deadline)
+      until shutdown_complete?
+        remaining = remaining_time(deadline)
+        break unless remaining.positive?
+
+        sleep([remaining, 0.01].min)
+      end
+    end
+
+    def shutdown_complete?
+      !alive? && !output_readers_alive? && !process_group_alive?
+    end
+
+    def process_group_alive?
+      Process.kill(0, -pid)
+      true
+    rescue Errno::ESRCH
+      false
+    rescue Errno::EPERM
+      true
+    end
+
+    def monotonic_time
+      Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    end
+
+    def remaining_time(deadline)
+      deadline - monotonic_time
+    end
+  end
+end
diff --git a/spec/support/framework_runtime_control_e2e_support.rb b/spec/support/framework_runtime_control_e2e_support.rb
index 4ccc12b7..4da22e58 100644
--- a/spec/support/framework_runtime_control_e2e_support.rb
+++ b/spec/support/framework_runtime_control_e2e_support.rb
@@ -7,11 +7,11 @@
 
 require 'json'
 require 'fileutils'
-require 'open3'
 require 'rbconfig'
 require 'securerandom'
 require 'timeout'
 require 'tmpdir'
+require_relative 'e2e_subprocess'
 
 module FrameworkRuntimeControlE2ESupport
   def framework_runtime_control_env(framework:, framework_gem_root:, database_url:, namespace:, worker_name:, worker: false)
@@ -165,11 +165,15 @@ def wait_for_framework_runtime_phase(state_file, *phases)
     end
   end
 
-  def wait_for_framework_runtime_start(state_file, wait_thr, stdout_and_stderr)
+  def wait_for_framework_runtime_start(state_file, process)
     wait_until do
-      if wait_thr.join(0)
-        output = stdout_and_stderr.read
-        raise "worker exited before runtime control started:\n#{output}"
+      unless process.alive?
+        begin
+          process.wait_for_output
+        rescue KaryaSpecSupport::E2ESubprocess::OutputTimeout
+          raise "worker exited before runtime control started and output remained open:\n#{process.output}"
+        end
+        raise "worker exited before runtime control started:\n#{process.output}"
       end
 
       next unless File.exist?(state_file)
@@ -204,18 +208,12 @@ def wait_until(timeout: 10)
     end
   end
 
-  def cleanup_framework_worker_process(wait_thr)
-    return unless wait_thr.alive?
-
-    Process.kill('TERM', wait_thr.pid)
-    Timeout.timeout(2) do
-      wait_thr.join
-    end
-    Process.kill('KILL', wait_thr.pid) if wait_thr.alive?
-  rescue Timeout::Error
-    Process.kill('KILL', wait_thr.pid) if wait_thr.alive?
-  rescue Errno::ESRCH
-    nil
+  def run_framework_runtime_command(framework:, action:, queue:, worker_name:, env:)
+    KaryaSpecSupport::E2ESubprocess.capture(
+      env,
+      *framework_runtime_command(framework:, action:, queue:, worker_name:),
+      chdir: current_app_root
+    )
   end
 
   def request_framework_force_stop(supervisor_pid:, draining_marker_path:)
@@ -312,71 +310,75 @@ def runtime_control_rakefile_source(framework:)
         framework_gem_root:,
         database_url:,
         namespace:,
-          worker_name:,
-          worker: true
+        worker_name:,
+        worker: true
       )
 
-      Open3.popen2e(
+      process = KaryaSpecSupport::E2ESubprocess.new(
         command_env,
         *framework_worker_command(framework:, queue:, worker_name:),
         chdir: current_app_root
-      ) do |_stdin, stdout_and_stderr, wait_thr|
-        begin
-          wait_for_framework_runtime_start(state_file, wait_thr, stdout_and_stderr)
-
-          inspect_stdout, inspect_stderr, inspect_status = Open3.capture3(
-            framework_runtime_control_env(
-              framework:,
-              framework_gem_root:,
-              database_url:,
-              namespace:,
-              worker_name:
-            ),
-            *framework_runtime_command(framework:, action: :inspect, queue:, worker_name:),
-            chdir: current_app_root
-          )
-          inspect_payload = parse_framework_runtime_json(inspect_stdout)
-
-          expect(inspect_status.exitstatus).to eq(0), -> { "stderr:\n#{inspect_stderr}" }
-          expect(inspect_payload.fetch('snapshot').fetch('phase')).to match(/\A(?:starting|running|draining)\z/)
-
-          drain_stdout, drain_stderr, drain_status = Open3.capture3(
-            framework_runtime_control_env(
-              framework:,
-              framework_gem_root:,
-              database_url:,
-              namespace:,
-              worker_name:
-            ),
-            *framework_runtime_command(framework:, action: :drain, queue:, worker_name:),
-            chdir: current_app_root
+      )
+      begin
+        wait_for_framework_runtime_start(state_file, process)
+
+        inspect_stdout, inspect_stderr, inspect_status = run_framework_runtime_command(
+          framework:,
+          action: :inspect,
+          queue:,
+          worker_name:,
+          env: framework_runtime_control_env(
+            framework:,
+            framework_gem_root:,
+            database_url:,
+            namespace:,
+            worker_name:
           )
+        )
+        inspect_payload = parse_framework_runtime_json(inspect_stdout)
+
+        expect(inspect_status.exitstatus).to eq(0), -> { "stderr:\n#{inspect_stderr}" }
+        expect(inspect_payload.fetch('snapshot').fetch('phase')).to match(/\A(?:starting|running|draining)\z/)
 
-          expect(drain_status.exitstatus).to eq(0), -> { "stdout:\n#{drain_stdout}\n\nstderr:\n#{drain_stderr}" }
-          expect(wait_for_framework_runtime_phase(state_file, 'draining')).to include('phase' => 'draining')
-
-          force_stop_stdout, force_stop_stderr, force_stop_status = Open3.capture3(
-            framework_runtime_control_env(
-              framework:,
-              framework_gem_root:,
-              database_url:,
-              namespace:,
-              worker_name:
-            ),
-            *framework_runtime_command(framework:, action: :force_stop, queue:, worker_name:),
-            chdir: current_app_root
+        drain_stdout, drain_stderr, drain_status = run_framework_runtime_command(
+          framework:,
+          action: :drain,
+          queue:,
+          worker_name:,
+          env: framework_runtime_control_env(
+            framework:,
+            framework_gem_root:,
+            database_url:,
+            namespace:,
+            worker_name:
           )
+        )
 
-          expect(force_stop_status.exitstatus).to eq(0), -> { "stdout:\n#{force_stop_stdout}\n\nstderr:\n#{force_stop_stderr}" }
-          runtime_phase = wait_for_framework_runtime_phase(state_file, 'force_stopping', 'stopped').fetch('phase')
-          expect(runtime_phase).to match(/\A(?:force_stopping|stopped)\z/)
-        rescue Timeout::Error
-          output = wait_thr.join(0) ? stdout_and_stderr.read : '(worker still running)'
-          state_payload = File.exist?(state_file) ? File.read(state_file) : '(missing state file)'
-          raise "worker runtime control timed out:\nstate:\n#{state_payload}\n\noutput:\n#{output}"
-        ensure
-          cleanup_framework_worker_process(wait_thr)
-        end
+        expect(drain_status.exitstatus).to eq(0), -> { "stdout:\n#{drain_stdout}\n\nstderr:\n#{drain_stderr}" }
+        expect(wait_for_framework_runtime_phase(state_file, 'draining')).to include('phase' => 'draining')
+
+        force_stop_stdout, force_stop_stderr, force_stop_status = run_framework_runtime_command(
+          framework:,
+          action: :force_stop,
+          queue:,
+          worker_name:,
+          env: framework_runtime_control_env(
+            framework:,
+            framework_gem_root:,
+            database_url:,
+            namespace:,
+            worker_name:
+          )
+        )
+
+        expect(force_stop_status.exitstatus).to eq(0), -> { "stdout:\n#{force_stop_stdout}\n\nstderr:\n#{force_stop_stderr}" }
+        runtime_phase = wait_for_framework_runtime_phase(state_file, 'force_stopping', 'stopped').fetch('phase')
+        expect(runtime_phase).to match(/\A(?:force_stopping|stopped)\z/)
+      rescue Timeout::Error
+        state_payload = File.exist?(state_file) ? File.read(state_file) : '(missing state file)'
+        raise "worker runtime control timed out:\nstate:\n#{state_payload}\n\noutput:\n#{process.output}"
+      ensure
+        KaryaSpecSupport::E2ESubprocess.close_preserving_failure(process)
       end
     end
   end
@@ -396,55 +398,55 @@ def runtime_control_rakefile_source(framework:)
         framework_gem_root:,
         database_url:,
         namespace:,
-          worker_name:,
-          worker: true
+        worker_name:,
+        worker: true
       )
 
-      Open3.popen2e(
+      process = KaryaSpecSupport::E2ESubprocess.new(
         worker_env,
         *framework_worker_command(framework:, queue:, worker_name:),
         chdir: current_app_root
-      ) do |_stdin, stdout_and_stderr, wait_thr|
-        begin
-          wait_for_framework_runtime_start(state_file, wait_thr, stdout_and_stderr)
-          write_stale_runtime_state_file!(live_state_file: state_file, stale_state_file:)
-
-          command_stdout, command_stderr, command_status = Open3.capture3(
-            framework_runtime_control_env(
-              framework:,
-              framework_gem_root:,
-              database_url:,
-              namespace:,
-              worker_name: stale_worker_name
-            ),
-            *framework_runtime_command(framework:, action: :drain, queue:, worker_name: stale_worker_name),
-            chdir: current_app_root
+      )
+      begin
+        wait_for_framework_runtime_start(state_file, process)
+        write_stale_runtime_state_file!(live_state_file: state_file, stale_state_file:)
+
+        command_stdout, command_stderr, command_status = run_framework_runtime_command(
+          framework:,
+          action: :drain,
+          queue:,
+          worker_name: stale_worker_name,
+          env: framework_runtime_control_env(
+            framework:,
+            framework_gem_root:,
+            database_url:,
+            namespace:,
+            worker_name: stale_worker_name
           )
-
-          combined_output = "#{command_stdout}\n#{command_stderr}"
-          expect(command_status.exitstatus).not_to eq(0), lambda {
-            stale_state_payload = File.exist?(stale_state_file) ? File.read(stale_state_file) : '(missing state file)'
-            <<~TEXT
-              expected stale-token runtime command to fail
-              stdout:
-              #{command_stdout}
-
-              stderr:
-              #{command_stderr}
-
-              state:
-              #{stale_state_payload}
-            TEXT
-          }
-          expect(combined_output).to include('runtime control token does not match the running supervisor')
-        rescue Timeout::Error
-          output = wait_thr.join(0) ? stdout_and_stderr.read : '(worker still running)'
-          state_payload = File.exist?(state_file) ? File.read(state_file) : '(missing state file)'
-          raise "worker runtime control timed out:\nstate:\n#{state_payload}\n\noutput:\n#{output}"
-        ensure
-          File.delete(stale_state_file) if File.exist?(stale_state_file)
-          cleanup_framework_worker_process(wait_thr)
-        end
+        )
+
+        combined_output = "#{command_stdout}\n#{command_stderr}"
+        expect(command_status.exitstatus).not_to eq(0), lambda {
+          stale_state_payload = File.exist?(stale_state_file) ? File.read(stale_state_file) : '(missing state file)'
+          <<~TEXT
+            expected stale-token runtime command to fail
+            stdout:
+            #{command_stdout}
+
+            stderr:
+            #{command_stderr}
+
+            state:
+            #{stale_state_payload}
+          TEXT
+        }
+        expect(combined_output).to include('runtime control token does not match the running supervisor')
+      rescue Timeout::Error
+        state_payload = File.exist?(state_file) ? File.read(state_file) : '(missing state file)'
+        raise "worker runtime control timed out:\nstate:\n#{state_payload}\n\noutput:\n#{process.output}"
+      ensure
+        File.delete(stale_state_file) if File.exist?(stale_state_file)
+        KaryaSpecSupport::E2ESubprocess.close_preserving_failure(process)
       end
     end
   end