diff --git a/fixtures/async/container/a_child.rb b/fixtures/async/container/a_child.rb
new file mode 100644
index 0000000..3487eb6
--- /dev/null
+++ b/fixtures/async/container/a_child.rb
@@ -0,0 +1,161 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+module Async
+	module Container
+		AChild = Sus::Shared("a child") do
+			def wait_for_child_started(input)
+				input.gets
+			end
+			
+			def start_ready_child
+				subject.call(name: "test-child") do |instance|
+					instance.ready!(status: "ready")
+				end
+			end
+			
+			def start_sleeping_child
+				::IO.pipe do |input, output|
+					child = subject.call(name: "test-child") do
+						output.puts "ready"
+						sleep
+					rescue Interrupt
+						# Graceful shutdown.
+					end
+					
+					wait_for_child_started(input)
+					
+					return child
+				end
+			end
+			
+			def start_uncooperative_child
+				::IO.pipe do |input, output|
+					child = subject.call(name: "test-child") do
+						ready_sent = false
+						
+						loop do
+							begin
+								unless ready_sent
+									output.puts "ready"
+									ready_sent = true
+								end
+								
+								sleep
+							rescue Interrupt
+								# Ignore graceful shutdown.
+							end
+						end
+					end
+					
+					wait_for_child_started(input)
+					
+					return child
+				end
+			end
+			
+			it "receives notifications and returns a successful status" do
+				child = start_ready_child
+				
+				messages = []
+				status = child.wait do |message|
+					messages << message
+				end
+				
+				expect(messages).to be == [{ready: true, status: "ready"}]
+				expect(status).to be(:success?)
+			end
+			
+			it "tracks when the child last sent a message" do
+				child = start_ready_child
+				last_updated_at = child.last_updated_at
+				
+				messages = []
+				child.wait do |message|
+					messages << message
+				end
+				
+				expect(messages).to be == [{ready: true, status: "ready"}]
+				expect(child.last_updated_at).to be >= last_updated_at
+			end
+			
+			it "can receive messages until the child is ready" do
+				child = subject.call(name: "test-child") do |instance|
+					instance.ready!(status: "ready")
+					sleep
+				rescue Interrupt
+					# Graceful shutdown.
+				end
+				
+				result = child.receive do
+					break :ready if child.ready?
+				end
+				
+				expect(result).to be == :ready
+				expect(child).to be(:ready?)
+				
+				status = child.stop(true)
+				
+				expect(status).to be(:success?)
+			end
+			
+			it "returns a successful status for normal exit" do
+				child = subject.call(name: "test-child") do
+					# Exit normally.
+				end
+				
+				status = child.wait
+				
+				expect(status).to be(:success?)
+			end
+			
+			it "returns nil when wait times out" do
+				child = start_sleeping_child
+				last_updated_at = child.last_updated_at
+				
+				expect(child.wait(0.001)).to be_nil
+				expect(child.last_updated_at).to be == last_updated_at
+				
+				status = child.stop(false)
+				
+				expect(status).not.to be(:success?)
+			end
+			
+			it "can stop gracefully" do
+				child = start_sleeping_child
+				
+				status = child.stop(true)
+				
+				expect(status).to be(:success?)
+			end
+			
+			it "can kill immediately" do
+				child = start_sleeping_child
+				
+				status = child.stop(false)
+				
+				expect(status).not.to be(:success?)
+			end
+			
+			it "kills the child when graceful shutdown times out" do
+				child = start_uncooperative_child
+				
+				status = child.stop(0.001)
+				
+				expect(status).not.to be(:success?)
+			end
+			
+			it "returns a failure status when the child fails" do
+				child = subject.call(name: "test-child") do
+					raise "boom"
+				end
+				
+				status = child.wait
+				
+				expect(status).not.to be(:success?)
+			end
+		end
+	end
+end
diff --git a/fixtures/async/container/a_container.rb b/fixtures/async/container/a_container.rb
index 207519c..56d20a5 100644
--- a/fixtures/async/container/a_container.rb
+++ b/fixtures/async/container/a_container.rb
@@ -1,529 +1,174 @@
 # frozen_string_literal: true
 
 # Released under the MIT License.
-# Copyright, 2019-2026, by Samuel Williams.
+# Copyright, 2026, by Samuel Williams.
 
-require "async"
+require "sus/shared"
+require "tmpdir"
+require "fileutils"
 
 module Async
 	module Container
 		AContainer = Sus::Shared("a container") do
-			let(:policy) {Async::Container::Policy::DEFAULT}
-			let(:container) {subject.new(policy: policy)}
+			let(:container) {subject.new}
 			
-			with "#run" do
-				it "can run several instances concurrently" do
-					container.run do
-						sleep(1)
-					end
-					
-					expect(container).to be(:running?)
-					
-					container.stop(true)
-					
-					expect(container).not.to be(:running?)
-				end
+			def temporary_directory
+				path = Dir.mktmpdir("async-container")
 				
-				it "can stop an uncooperative child process" do
-					container.run do
-						while true
-							begin
-								sleep(1)
-							rescue Interrupt
-								# Ignore.
-							end
-						end
-					end
-					
-					expect(container).to be(:running?)
-					
-					# TODO Investigate why without this, the interrupt can occur before the process is sleeping...
-					sleep 0.001
-					
-					container.stop(true)
-					
-					expect(container).not.to be(:running?)
+				begin
+					yield path
+				ensure
+					FileUtils.remove_entry(path)
 				end
 			end
 			
-			with "#async" do
-				it "can run concurrently" do
-					input, output = IO.pipe
-					
-					container.async do
-						output.write "Hello World"
-					end
-					
-					container.wait
-					
-					output.close
-					expect(input.read).to be == "Hello World"
-				end
-				
-				it "can run concurrently" do
-					container.async(name: "Sleepy Jerry") do |task, instance|
-						3.times do |i|
-							instance.name = "Counting Sheep #{i}"
-							
-							sleep 0.01
-						end
-					end
-					
-					container.wait
+			def append_line(path, line)
+				File.open(path, "a") do |file|
+					file.flock(File::LOCK_EX)
+					file.puts(line)
 				end
 			end
 			
-			with "Async{}" do
-				it "can wait inside an Async task after spawning outside Async" do
-					input, output = IO.pipe
-					
-					container.spawn do
-						output.write(".")
-					end
-					
-					Async do
-						container.wait
-					end.wait
-					
-					output.close
-					expect(input.read).to be == "."
-				end
-				
-				it "can spawn and wait inside the same Async task" do
-					input, output = IO.pipe
-					
-					Async do
-						container.spawn do
-							output.write(".")
-						end
-						
-						container.wait
-					end.wait
-					
-					output.close
-					expect(input.read).to be == "."
-				end
-				
-				it "can wait while a health monitor is active" do
-					container.spawn(health_check_timeout: 10.0) do |instance|
-						instance.ready!
-					end
-					
-					Async do |task|
-						task.with_timeout(2.0) do
-							container.wait
-						end
-					end.wait
-					
-					expect(container.statistics).to have_attributes(failures: be == 0)
-				end
-				
-				it "can start, wait until ready, and stop inside Sync" do
-					Sync do
-						begin
-							2.times do |i|
-								container.spawn(name: "worker #{i}") do |instance|
-									instance.ready!
-									sleep
-								end
-							end
-							
-							container.wait_until_ready
-							
-							expect(container.group.running).to have_attributes(size: be == 2)
-						ensure
-							container.stop if container.running?
-						end
-					end
-				end
+			def read_lines(path)
+				File.exist?(path) ? File.readlines(path, chomp: true) : []
 			end
 			
-			it "should be blocking" do
-				skip "Fiber.blocking? is not supported!" unless Fiber.respond_to?(:blocking?)
-				
-				input, output = IO.pipe
-				
-				container.spawn do
-					output.write(Fiber.blocking? != false)
+			it "spawns a child and observes readiness" do
+				container.spawn(name: "worker") do |instance|
+					instance.ready!(status: "ready")
+					sleep
 				end
 				
-				container.wait
+				expect(container.wait_until_ready).to be == true
+				expect(container).to be(:running?)
+				expect(container.state.values.any?{|state| state[:ready] == true && state[:status] == "ready"}).to be == true
+				
+				container.stop(false)
 				
-				output.close
-				expect(input.read).to be == "true"
+				expect(container).not.to be(:running?)
 			end
 			
-			with "instance" do
-				it "can generate JSON representation" do
-					IO.pipe do |input, output|
-						container.spawn do |instance|
-							output.write(instance.to_json)
-						end
-						
-						container.wait
-						
-						expect(container.statistics).to have_attributes(failures: be == 0)
-						
-						output.close
-						instance = JSON.parse(input.read, symbolize_names: true)
-						expect(instance).to have_keys(
-							process_id: be_a(Integer),
-							name: be_a(String),
-						)
-					end
-				end
-				
-				it "can exec with ready: true without premature termination" do
-					container.spawn(restart: false) do |instance|
-						# Using exec with ready: true should not cause the process to be killed
-						# by hang prevention, even though the notification pipe stays open.
-						instance.exec("sleep", "1", ready: true)
-					end
+			it "runs the requested number of workers" do
+				temporary_directory do |directory|
+					path = File.join(directory, "workers.log")
 					
-					# Wait for the process to become ready:
-					container.wait_until_ready
-					
-					# Sleep longer than the hang prevention timeout (0.1s) to verify
-					# the process isn't prematurely killed:
-					sleep(0.2)
-					
-					# The process should still be running (not killed by hang prevention):
-					expect(container).to be(:running?)
+					container.run(count: 2) do |instance|
+						append_line(path, "ready")
+						instance.ready!
+						sleep
+					end
 					
-					# Now stop the container:
+					expect(container.wait_until_ready).to be == true
+					expect(read_lines(path).size).to be == 2
+				ensure
 					container.stop(false)
 				end
 			end
 			
-			with "#sleep" do
-				it "can sleep for a short time" do
-					container.spawn do
-						sleep(0.01)
-						raise "Boom"
-					end
+			it "runs asynchronous work" do
+				temporary_directory do |directory|
+					path = File.join(directory, "async.log")
 					
-					expect(container.statistics).to have_attributes(failures: be == 0)
+					container.async do
+						append_line(path, "done")
+					end
 					
 					container.wait
 					
-					expect(container.statistics).to have_attributes(failures: be == 1)
+					expect(read_lines(path)).to be == ["done"]
+					expect(container.statistics.failures).to be == 0
 				end
 			end
 			
-			with "#stop" do
-				let(:policy) do
-					Class.new(Async::Container::Policy) do
-						def initialize
-							@events = []
-						end
-						
-						attr :events
-						
-						def child_exit(container, child, status, name:, key:, **options)
-							# Capture the state of `stopping?`:
-							@events << [:child_exit, container.stopping?]
-						end
-					end.new
+			it "waits for children to exit normally" do
+				container.spawn do |instance|
+					instance.ready!
 				end
 				
-				it "can gracefully stop the child process" do
-					container.spawn do
-						sleep(1)
-					rescue Interrupt
-						# Ignore.
-					end
-					
-					expect(container).to be(:running?)
-					expect(container).not.to be(:stopping?)
-					
-					# See above.
-					sleep 0.001
-					
-					container.stop(true)
-					
-					expect(container).not.to be(:running?)
-					expect(policy.events).to be == [[:child_exit, true]]
-				end
-				
-				it "can forcefully stop the child process" do
-					container.spawn do
-						sleep(1)
-					rescue Interrupt
-						# Ignore.
-					end
-					
-					expect(container).to be(:running?)
-					
-					# See above.
-					sleep 0.001
-					
-					container.stop(false)
-					
-					expect(container).not.to be(:running?)
-					expect(policy.events).to be == [[:child_exit, true]]
-				end
+				container.wait
 				
-				it "can stop an uncooperative child process" do
-					container.spawn do
-						while true
-							begin
-								sleep(1)
-							rescue Interrupt
-								# Ignore.
-							end
-						end
-					end
-					
-					expect(container).to be(:running?)
-					
-					# See above.
-					sleep 0.001
-					
-					container.stop(true)
-					
-					expect(container).not.to be(:running?)
-				end
+				expect(container).not.to be(:running?)
+				expect(container.statistics).to have_attributes(spawns: be == 1, failures: be == 0)
+				expect(container).not.to be(:failed?)
 			end
 			
-			with "#ready" do
-				it "can notify the ready pipe in an asynchronous context" do
-					container.run do |instance|
-						Async do
-							instance.ready!
-						end
-					end
-					
-					expect(container).to be(:running?)
-					
-					container.wait
-					
-					container.stop
-					
-					expect(container).not.to be(:running?)
+			it "records child failures" do
+				container.spawn do
+					raise "boom"
 				end
+				
+				container.wait
+				
+				expect(container).not.to be(:running?)
+				expect(container.statistics).to have_attributes(spawns: be == 1, failures: be == 1)
+				expect(container).to be(:failed?)
 			end
 			
-			with "health_check_timeout:" do
-				let(:container) {subject.new(health_check_interval: 1.0)}
-				
-				it "should not terminate a child process if it updates its state within the specified time" do
-					# We use #run here to hit the Hybrid container code path:
-					container.run(count: 1, health_check_timeout: 1.0) do |instance|
-						instance.ready!
-						
-						10.times do
-							instance.ready!
-							sleep(0.5)
-						end
-					end
-					
-					container.wait
-					
-					expect(container.statistics).to have_attributes(failures: be == 0)
+			it "can stop gracefully" do
+				container.spawn do |instance|
+					instance.ready!
+					sleep
+				rescue Interrupt
+					# Graceful shutdown.
 				end
 				
-				it "can terminate a child process if it does not update its state within the specified time" do
-					container.spawn(health_check_timeout: 1.0) do |instance|
-						instance.ready!
-						
-						# This should trigger the health check - since restart is false, the process will be terminated:
-						sleep
-					end
-					
-					container.wait
-					
-					expect(container.statistics).to have_attributes(failures: be > 0)
-				end
+				expect(container.wait_until_ready).to be == true
 				
-				it "can kill a child process even if it ignores exceptions/signals" do
-					# This process never calls ready!, so we need startup_timeout to kill it
-					container.spawn(health_check_timeout: 1.0, startup_timeout: 1.0) do |instance|
-						while true
-							begin
-								sleep 1
-							rescue Exception => error
-								# Ignore.
-							end
-						end
-					end
-					
-					container.wait
-					
-					expect(container.statistics).to have_attributes(failures: be > 0)
-				end
+				container.stop(true)
+				
+				expect(container).not.to be(:running?)
+				expect(container.statistics.failures).to be == 0
 			end
 			
-			with "startup_timeout:" do
-				let(:container) {subject.new(health_check_interval: 1.0)}
-				
-				it "should not terminate a child process if it becomes ready within the startup timeout" do
-					container.spawn(startup_timeout: 2.0) do |instance|
-						instance.status!("Starting...")
-						sleep(0.5)
-						
-						instance.status!("Preparing...")
-						sleep(0.5)
-						
-						instance.ready!
-						
-						# Keep running
-						sleep(1)
-					end
-					
-					container.wait
-					
-					expect(container.statistics).to have_attributes(failures: be == 0)
+			it "can stop immediately" do
+				container.spawn do |instance|
+					instance.ready!
+					sleep
 				end
 				
-				it "can terminate a child process if it does not become ready within the startup timeout" do
-					container.spawn(startup_timeout: 1.0) do |instance|
-						instance.status!("Starting...")
-						
-						# Never call ready! - should be killed by startup timeout
-						sleep
-					end
-					
-					container.wait
-					
-					expect(container.statistics).to have_attributes(failures: be > 0)
-				end
+				expect(container.wait_until_ready).to be == true
 				
-				it "can terminate a child process that sends status messages but never becomes ready" do
-					container.spawn(startup_timeout: 1.0) do |instance|
-						# Send status messages but never become ready
-						while true
-							instance.status!("Still starting...")
-							sleep(0.3)
-						end
-					end
-					
-					container.wait
-					
-					expect(container.statistics).to have_attributes(failures: be > 0)
-				end
+				container.stop(false)
 				
-				it "transitions from startup timeout to health check timeout after becoming ready" do
-					container.spawn(startup_timeout: 2.0, health_check_timeout: 1.0) do |instance|
-						instance.status!("Starting...")
-						sleep(0.5)
-						
-						instance.ready!
-						
-						# After becoming ready, health_check_timeout should apply
-						# Don't send any more messages - should be killed by health check timeout
-						sleep
-					end
-					
-					container.wait
-					
-					expect(container.statistics).to have_attributes(failures: be > 0)
-				end
-				
-				it "resets the clock when the child becomes ready" do
-					container.spawn(startup_timeout: 1.5, health_check_timeout: 1.0) do |instance|
-						instance.status!("Starting...")
-						sleep(1.0) # Use up most of startup timeout
-						
-						instance.ready! # Clock should reset here
-						
-						# After ready, health_check_timeout applies (1.0 seconds)
-						# Send ready! messages periodically to stay alive
-						5.times do
-							sleep(0.4)
-							instance.ready!
-						end
-					end
-					
-					container.wait
-					
-					expect(container.statistics).to have_attributes(failures: be == 0)
-				end
+				expect(container).not.to be(:running?)
 			end
 			
-			with "broken children" do
-				it "can handle children that ignore termination with SIGKILL fallback" do
-					# Test behavior that works for both processes (signals) and threads (exceptions)
-					container.spawn(restart: false) do |instance|
-						instance.ready!
-						
-						# Ignore termination attempts in a way appropriate to the container type
-						if container.class.multiprocess?
-							# For multiprocess containers - ignore signals
-							Signal.trap(:INT){}
-							Signal.trap(:TERM){}
-							while true
-								sleep(0.1)
-							end
-						else
-							# For threaded containers - ignore exceptions
-							while true
-								begin
-									sleep(0.1)
-								rescue Async::Container::Interrupt, Async::Container::Terminate
-									# Ignore termination attempts
-								end
-							end
-						end
-					end
-					
-					container.wait_until_ready
-					
-					# Try to stop with a very short timeout to force escalation
-					start_time = Time.now
-					container.stop(0.1) # Very short timeout
-					end_time = Time.now
+			it "restarts failed children when requested" do
+				temporary_directory do |directory|
+					marker_path = File.join(directory, "started")
 					
-					# Should stop successfully via SIGKILL/thread termination
-					expect(container.size).to be == 0
-					
-					# Should not hang - escalation should work
-					expect(end_time - start_time).to be < 2.0
-				end
-				
-				it "can handle unresponsive children that close pipes but don't exit" do
-					container.spawn(restart: false) do |instance|
-						instance.ready!
-						
-						# Close communication pipe to simulate hung process:
-						begin
-							if instance.respond_to?(:out)
-								instance.out.close if instance.out && !instance.out.closed?
-							end
-						rescue
-							# Ignore close errors.
-						end
-						
-						# Become unresponsive:
-						if container.class.multiprocess?
-							# For multiprocess containers - ignore signals and close file descriptors:
-							Signal.trap(:INT){}
-							Signal.trap(:TERM){}
-							(4..256).each do |fd|
-								begin
-									IO.for_fd(fd).close
-								rescue
-									# Ignore errors
-								end
-							end
-							loop {} # Tight loop
+					container.spawn(restart: true) do |instance|
+						if File.exist?(marker_path)
+							instance.ready!
+							sleep
 						else
-							# For threaded containers - just become unresponsive
-							loop{} # Tight loop, no exception handling
+							File.write(marker_path, "started")
+							raise "boom"
 						end
 					end
 					
-					container.wait_until_ready
+					expect(container.wait_until_ready).to be == true
 					
-					# Should not hang even with unresponsive children
-					start_time = Time.now
-					container.stop(1.0)
-					end_time = Time.now
+					container.stop(false)
 					
-					expect(container.size).to be == 0
-					# Should complete reasonably quickly via hang prevention
-					expect(end_time - start_time).to be < 5.0
+					expect(container.statistics).to have_attributes(spawns: be == 2, restarts: be == 1, failures: be == 1)
+				ensure
+					container.stop(false)
 				end
 			end
+			
+			it "does not spawn a duplicate keyed child" do
+				expect(container.spawn(key: :worker) do |instance|
+					instance.ready!
+					sleep
+				end).to be == true
+				
+				expect(container.wait_until_ready).to be == true
+				expect(container.spawn(key: :worker){sleep}).to be == false
+				expect(container[:worker]).not.to be_nil
+				
+				container.stop(false)
+			end
 		end
 	end
 end
diff --git a/lib/async/container/channel.rb b/lib/async/container/channel.rb
index d207442..0d60ea6 100644
--- a/lib/async/container/channel.rb
+++ b/lib/async/container/channel.rb
@@ -10,9 +10,9 @@ module Container
 		# Provides a basic multi-thread/multi-process uni-directional communication channel.
 		class Channel
 			# Initialize the channel using a pipe.
-			def initialize(timeout: 1.0)
+			def initialize(timeout: nil)
 				@in, @out = ::IO.pipe
-				@in.timeout = timeout
+				@in.timeout = timeout if timeout
 			end
 			
 			# The input end of the pipe.
@@ -46,8 +46,7 @@ def receive
 				if data = @in.gets
 					return JSON.parse(data, symbolize_names: true)
 				end
-			rescue => error
-				Console.error(self, "Error during channel receive!", error)
+			rescue JSON::ParserError
 				return nil
 			end
 		end
diff --git a/lib/async/container/child.rb b/lib/async/container/child.rb
new file mode 100644
index 0000000..e8cc3af
--- /dev/null
+++ b/lib/async/container/child.rb
@@ -0,0 +1,175 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require "async/deadline"
+
+require_relative "channel"
+require_relative "error"
+
+module Async
+	module Container
+		# Represents a child execution unit managed by a container.
+		class Child
+			# The default amount of time to wait for a child to be reaped after its
+			# notification channel closes.
+			REAP_TIMEOUT = 1.0
+			
+			# Initialize the child with the given notification channel.
+			# @parameter channel [Channel] The channel used to receive child notifications.
+			# @parameter name [String | Nil] The optional child name.
+			# @parameter options [Hash] Additional child options.
+			def initialize(channel, name: nil, **options)
+				@channel = channel
+				@name = name
+				@options = options
+				@state = {}
+				@last_updated_at = Clock.now
+			end
+			
+			# @attribute [Channel] The channel for the child.
+			attr :channel
+			
+			# @attribute [String] The name for the child.
+			attr :name
+			
+			# @attribute [Hash] The current child state, derived from notification messages.
+			attr :state
+			
+			# @attribute [Float] The last time the child sent a message.
+			attr :last_updated_at
+			
+			# Whether the child has reported readiness.
+			def ready?
+				@state[:ready] == true
+			end
+			
+			# Update the child state from a notification message.
+			def update(message)
+				@state.update(message)
+				@last_updated_at = Clock.now
+			end
+			
+			# Stop the child with a multi-phase shutdown sequence.
+			#
+			# A graceful shutdown sends an interrupt first and waits up to `graceful`
+			# seconds. If the child is still running, or if graceful shutdown is
+			# disabled, the child is killed and waited for indefinitely.
+			#
+			# @parameter graceful [Boolean | Numeric] Whether to send an interrupt first or skip directly to kill.
+			def stop(graceful = true, &block)
+				status = nil
+				
+				if graceful
+					self.interrupt!
+					
+					timeout = (graceful == true) ? nil : graceful
+					status = self.wait(timeout, &block)
+				end
+				
+				return status if status
+			ensure
+				unless status
+					self.kill!
+					status = self.wait(nil, &block)
+				end
+				
+				return status
+			end
+			
+			# Receive notification messages from the child until a message is available, the channel closes, or the timeout expires.
+			# @parameter timeout [Numeric | Nil] The maximum time to wait for a message.
+			# @yields {|message| ...} Each received notification message.
+			# 	@parameter message [Hash] The notification message from the child.
+			# @returns [Hash | Boolean | Nil] The message, `false` on timeout, or `nil` when the channel closes.
+			def receive(timeout = nil, &block)
+				deadline = Deadline.new(timeout) if timeout
+				
+				while true
+					if timeout
+						if deadline.expired?
+							return false
+						end
+						
+						unless @channel.in.wait_readable(deadline.remaining)
+							return false
+						end
+					else
+						@channel.in.wait_readable
+					end
+					
+					if message = @channel.receive
+						self.update(message)
+						if block_given?
+							yield message
+						else
+							return message
+						end
+					else
+						return nil
+					end
+				end
+			end
+			
+			# Drain notification messages until the channel closes or the timeout expires.
+			#
+			# @parameter timeout [Numeric | Nil] Maximum time to wait before returning `nil`.
+			# @returns [Object | Nil] The child status if the child exited, otherwise `nil` on timeout.
+			def wait(timeout = nil, &block)
+				deadline = Deadline.new(timeout) if timeout
+				
+				result = if block
+					receive(timeout, &block)
+				else
+					receive(timeout) do
+						# Drain messages until the child exits.
+					end
+				end
+				
+				case result
+				when false
+					return nil
+				when nil
+					timeout = deadline&.remaining || REAP_TIMEOUT
+					status = self.reap(timeout)
+					
+					unless status
+						self.kill!
+						status = self.reap
+					end
+					
+					return status
+				else
+					# The block broke early, so the child may still be running.
+					return nil
+				end
+			end
+			
+			# Reap the child after the channel has closed.
+			def reap(timeout = nil)
+				raise NotImplementedError
+			end
+			
+			# Interrupt the child, initiating graceful shutdown.
+			def interrupt!
+				raise NotImplementedError
+			end
+			
+			# Terminate the child.
+			def terminate!
+				raise NotImplementedError
+			end
+			
+			# Kill the child.
+			def kill!
+				raise NotImplementedError
+			end
+			
+			# Restart the child.
+			def restart!
+				raise NotImplementedError
+			end
+		end
+	end
+end
diff --git a/lib/async/container/controller.rb b/lib/async/container/controller.rb
index 4accfc8..3e5427d 100644
--- a/lib/async/container/controller.rb
+++ b/lib/async/container/controller.rb
@@ -10,8 +10,12 @@
 require_relative "notify"
 require_relative "policy"
 
+require "async"
+
 # This sets up graceful handling of SIGINT and SIGTERM.
+require "async/signals"
 require "async/signals/graceful"
+require "async/signals/handlers"
 
 module Async
 	module Container
@@ -44,10 +48,18 @@ def initialize(notify: Notify.open!, container_class: Container, graceful_stop:
 				@graceful_stop = graceful_stop
 				
 				@container = nil
-				@signals = {}
+				@handlers = Async::Signals::Handlers.new
 				
-				self.trap(SIGHUP) do
-					self.restart
+				@handlers.trap(SIGHUP) do |_signal, context|
+					context.raise(Restart)
+				end
+				
+				@handlers.trap(SIGINT) do |_signal, context|
+					context.raise(Interrupt)
+				end
+				
+				@handlers.trap(SIGTERM) do |_signal, context|
+					context.raise(Interrupt)
 				end
 			end
 			
@@ -83,7 +95,7 @@ def to_s
 			# @parameters signal [Symbol] The signal to trap, e.g. `:INT`.
 			# @parameters block [Proc] The signal handler to invoke.
 			def trap(signal, &block)
-				@signals[signal] = block
+				@handlers.trap(signal, &block)
 			end
 			
 			# Create a policy for managing child lifecycle events.
@@ -195,64 +207,30 @@ def restart
 			end
 			
 			# Enter the controller run loop, trapping `SIGINT` and `SIGTERM`.
-			def run
+			def run(signals: Async::Signals.default)
 				@notify&.status!("Initializing controller...")
 				
-				with_signal_handlers do
-					self.start
-					
-					while @container&.running?
-						begin
-							@container.wait
-						rescue SignalException => exception
-							if handler = @signals[exception.signo]
-								begin
-									handler.call
-								rescue SetupError => error
-									Console.error(self, error)
+				begin
+					Sync do |task|
+						self.start
+						
+						while @container&.running?
+							begin
+								signals.install(@handlers) do
+									@container.wait
 								end
-							else
-								raise
+							rescue Restart
+								self.restart
+							rescue Interrupt
+								self.stop(@graceful_stop)
 							end
 						end
 					end
+				rescue Interrupt
+					self.stop(@graceful_stop)
+				ensure
+					self.stop(false)
 				end
-			rescue Interrupt
-				self.stop
-			rescue Terminate
-				self.stop(false)
-			ensure
-				self.stop(false)
-			end
-			
-			private def with_signal_handlers
-				# I thought this was the default... but it doesn't always raise an exception unless you do this explicitly.
-				
-				interrupt_action = Signal.trap(:INT) do
-					# We use `Thread.current.raise(...)` so that exceptions are filtered through `Thread.handle_interrupt` correctly.
-					# $stderr.puts "Received INT signal, interrupting...", caller
-					::Thread.current.raise(Interrupt)
-				end
-				
-				# SIGTERM behaves the same as SIGINT by default.
-				terminate_action = Signal.trap(:TERM) do
-					# $stderr.puts "Received TERM signal, interrupting...", caller
-					::Thread.current.raise(Interrupt)  # Same as SIGINT
-				end
-				
-				hangup_action = Signal.trap(:HUP) do
-					# $stderr.puts "Received HUP signal, restarting...", caller
-					::Thread.current.raise(Restart)
-				end
-				
-				::Thread.handle_interrupt(SignalException => :never) do
-					yield
-				end
-			ensure
-				# Restore the interrupt handler:
-				Signal.trap(:INT, interrupt_action)
-				Signal.trap(:TERM, terminate_action)
-				Signal.trap(:HUP, hangup_action)
 			end
 		end
 	end
diff --git a/lib/async/container/forked.rb b/lib/async/container/forked.rb
index 21fe621..949102f 100644
--- a/lib/async/container/forked.rb
+++ b/lib/async/container/forked.rb
@@ -3,273 +3,25 @@
 # Released under the MIT License.
 # Copyright, 2017-2026, by Samuel Williams.
 
-require_relative "error"
-
 require_relative "generic"
-require_relative "channel"
-require_relative "notify/pipe"
+require_relative "forked/child"
 
 module Async
 	module Container
-		# A multi-process container which uses {Process.fork}.
-		class Forked < Generic
-			# Indicates that this is a multi-process container.
-			def self.multiprocess?
-				true
+		# Public factory for multi-process containers.
+		module Forked
+			# Create a generic container which spawns forked children.
+			# @parameter arguments [Array] Positional arguments for {Generic#initialize}.
+			# @parameter options [Hash] Keyword options for {Generic#initialize}.
+			# @returns [Generic] A generic container configured with {Forked::Child}.
+			def self.new(*arguments, **options)
+				Generic.new(Child, *arguments, **options)
 			end
 			
-			# Represents a running child process from the point of view of the parent container.
-			class Child < Channel
-				# Represents a running child process from the point of view of the child process.
-				class Instance < Notify::Pipe
-					# Wrap an instance around the {Process} instance from within the forked child.
-					# @parameter process [Process] The process intance to wrap.
-					def self.for(process)
-						instance = self.new(process.out)
-						
-						# The child process won't be reading from the channel:
-						process.close_read
-						
-						instance.name = process.name
-						
-						return instance
-					end
-					
-					# Initialize the child process instance.
-					#
-					# @parameter io [IO] The IO object to use for communication.
-					def initialize(io)
-						super
-						
-						@name = nil
-					end
-					
-					# Generate a hash representation of the process.
-					#
-					# @returns [Hash] The process as a hash, including `process_id` and `name`.
-					def as_json(...)
-						{
-							process_id: ::Process.pid,
-							name: @name,
-						}
-					end
-					
-					# Generate a JSON representation of the process.
-					#
-					# @returns [String] The process as JSON.
-					def to_json(...)
-						as_json.to_json(...)
-					end
-					
-					# Set the process title to the specified value.
-					#
-					# @parameter value [String] The name of the process.
-					def name= value
-						@name = value
-						
-						# This sets the process title to an empty string if the name is nil:
-						::Process.setproctitle(@name.to_s)
-					end
-					
-					# @returns [String] The name of the process.
-					def name
-						@name
-					end
-					
-					# Replace the current child process with a different one. Forwards arguments and options to {::Process.exec}.
-					# This method replaces the child process with the new executable, thus this method never returns.
-					#
-					# @parameter arguments [Array] The arguments to pass to the new process.
-					# @parameter ready [Boolean] If true, informs the parent process that the child is ready before exec. The notification pipe will still be passed to the exec'd process to prevent premature termination.
-					# @parameter options [Hash] Additional options to pass to {::Process.exec}.
-					def exec(*arguments, ready: true, **options)
-						# Always set up the notification pipe to be inherited by the exec'd process.
-						# This prevents the pipe from closing, which would trigger hang prevention and SIGKILL.
-						self.before_spawn(arguments, options)
-						
-						if ready
-							self.ready!(status: "(exec)")
-						end
-						
-						::Process.exec(*arguments, **options)
-					end
-				end
-				
-				# Fork a child process appropriate for a container.
-				#
-				# @returns [Process]
-				def self.fork(**options)
-					# $stderr.puts fork: caller
-					self.new(**options) do |process|
-						# Fork from `Thread.new` so the child does not inherit the parent fiber scheduler or the current caller's fiber stack. Only this short-lived thread is copied into the child process:
-						::Thread.new do
-							::Process.fork do
-								# We use `Thread.current.raise(...)` so that exceptions are filtered through `Thread.handle_interrupt` correctly.
-								Signal.trap(:INT){::Thread.current.raise(Interrupt)}
-								Signal.trap(:TERM){::Thread.current.raise(Interrupt)}  # Same as SIGINT.
-								Signal.trap(:HUP){::Thread.current.raise(Restart)}
-								
-								# Reset `SignalException` delivery because CRuby inherits the `Thread.handle_interrupt` mask stack across `Thread.new`. Async deliberately masks signal exceptions, and the signal traps above should be delivered promptly:
-								::Thread.handle_interrupt(SignalException => :immediate) do
-									yield Instance.for(process)
-								rescue Interrupt
-									# Graceful exit.
-								rescue Exception => error
-									Console.error(self, error)
-									
-									exit!(1)
-								end
-							end
-						end.value
-					end
-				end
-				
-				# Spawn a child process using {::Process.spawn}.
-				#
-				# The child process will need to inform the parent process that it is ready using a notification protocol.
-				#
-				# @parameter arguments [Array] The arguments to pass to the new process.
-				# @parameter name [String] The name of the process.
-				# @parameter options [Hash] Additional options to pass to {::Process.spawn}.
-				def self.spawn(*arguments, name: nil, **options)
-					self.new(name: name) do |process|
-						Notify::Pipe.new(process.out).before_spawn(arguments, options)
-						
-						::Process.spawn(*arguments, **options)
-					end
-				end
-				
-				# Initialize the process.
-				# @parameter name [String] The name to use for the child process.
-				def initialize(name: nil, **options)
-					super(**options)
-					
-					@name = name
-					@status = nil
-					@pid = nil
-					
-					@pid = yield(self)
-					
-					# The parent process won't be writing to the channel:
-					self.close_write
-				end
-				
-				# Convert the child process to a hash, suitable for serialization.
-				#
-				# @returns [Hash] The request as a hash.
-				def as_json(...)
-					{
-						name: @name,
-						pid: @pid,
-						status: @status&.to_i,
-					}
-				end
-				
-				# Convert the request to JSON.
-				#
-				# @returns [String] The request as JSON.
-				def to_json(...)
-					as_json.to_json(...)
-				end
-				
-				# Set the name of the process.
-				# Invokes {::Process.setproctitle} if invoked in the child process.
-				def name= value
-					@name = value
-					
-					# If we are the child process:
-					::Process.setproctitle(@name) if @pid.nil?
-				end
-				
-				# The name of the process.
-				# @attribute [String]
-				attr :name
-				
-				# @attribute [Integer] The process identifier.
-				attr :pid
-				
-				# A human readable representation of the process.
-				# @returns [String]
-				def inspect
-					"\#<#{self.class} name=#{@name.inspect} status=#{@status.inspect} pid=#{@pid.inspect}>"
-				end
-				
-				# @returns [String] A string representation of the process.
-				alias to_s inspect
-				
-				# Invoke {#terminate!} and then {#wait} for the child process to exit.
-				def close
-					self.terminate!
-					self.wait
-				ensure
-					super
-				end
-				
-				# Send `SIGINT` to the child process.
-				def interrupt!
-					unless @status
-						::Process.kill(:INT, @pid)
-					end
-				end
-				
-				# Send `SIGTERM` to the child process.
-				def terminate!
-					unless @status
-						::Process.kill(:TERM, @pid)
-					end
-				end
-				
-				# Send `SIGKILL` to the child process.
-				def kill!
-					unless @status
-						::Process.kill(:KILL, @pid)
-					end
-				end
-				
-				# Send `SIGHUP` to the child process.
-				def restart!
-					unless @status
-						::Process.kill(:HUP, @pid)
-					end
-				end
-				
-				# Wait for the child process to exit.
-				# @asynchronous This method may block.
-				#
-				# @parameter timeout [Numeric | Nil] Maximum time to wait before forceful termination.
-				# @returns [::Process::Status] The process exit status.
-				def wait(timeout = 0.1)
-					if @pid && @status.nil?
-						Console.debug(self, "Waiting for process to exit...", child: {process_id: @pid}, timeout: timeout)
-						
-						_, @status = ::Process.wait2(@pid, ::Process::WNOHANG)
-						
-						if @status.nil?
-							sleep(timeout) if timeout
-							
-							_, @status = ::Process.wait2(@pid, ::Process::WNOHANG)
-							
-							if @status.nil?
-								Console.warn(self, "Process is blocking, sending kill signal...", child: {process_id: @pid}, timeout: timeout)
-								self.kill!
-								
-								# Wait for the process to exit:
-								_, @status = ::Process.wait2(@pid)
-							end
-						end
-					end
-					
-					Console.debug(self, "Process exited.", child: {process_id: @pid, status: @status})
-					
-					return @status
-				end
-			end
-			
-			# Start a named child process and execute the provided block in it.
-			# @parameter name [String] The name (title) of the child process.
-			# @parameter block [Proc] The block to execute in the child process.
-			def start(name, &block)
-				Child.fork(name: name, &block)
+			# Whether this container uses multiple processes.
+			# @returns [Boolean]
+			def self.multiprocess?
+				true
 			end
 		end
 	end
diff --git a/lib/async/container/forked/child.rb b/lib/async/container/forked/child.rb
new file mode 100644
index 0000000..ef603a8
--- /dev/null
+++ b/lib/async/container/forked/child.rb
@@ -0,0 +1,121 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require_relative "../child"
+require_relative "instance"
+
+module Async
+	module Container
+		module Forked
+			# Represents a child process managed by a forked container.
+			class Child < Container::Child
+				# Start a child process using `fork`.
+				# @parameter channel [Channel] The notification channel for the child.
+				# @parameter name [String | Nil] The optional child name.
+				# @parameter options [Hash] Additional child options.
+				# @yields {|instance| ...} The child process body.
+				# 	@parameter instance [Instance] The child-side instance interface.
+				# @returns [Child] The forked child process.
+				def self.call(channel: Channel.new, name: nil, **options, &block)
+					process_id = ::Thread.new do
+						::Process.fork do
+							begin
+								Signal.trap(:INT){::Thread.current.raise(Interrupt)}
+								Signal.trap(:TERM){::Thread.current.raise(Interrupt)}
+								Signal.trap(:HUP){::Thread.current.raise(Restart)}
+								
+								::Thread.handle_interrupt(SignalException => :immediate) do
+									yield Instance.for(channel, name: name)
+								end
+							rescue Interrupt
+								# Graceful shutdown.
+							rescue Restart
+								# Graceful restart.
+							rescue Exception
+								exit!(1)
+							ensure
+								channel.close_write unless channel.out.closed?
+							end
+						end
+					end.value
+					
+					# The parent process won't be writing to the channel:
+					channel.close_write
+					
+					return self.new(process_id, channel, name: name, **options)
+				end
+				
+				# Initialize the child process wrapper.
+				# @parameter process_id [Integer] The process identifier.
+				# @parameter channel [Channel] The notification channel for the child.
+				# @parameter options [Hash] Additional child options.
+				def initialize(process_id, channel, **options)
+					@process_id = process_id
+					@status = nil
+					
+					super(channel, **options)
+				end
+				
+				# The process identifier.
+				# @attribute [Integer]
+				attr :process_id
+				
+				# Send `SIGINT` to the child process.
+				def interrupt!
+					unless @status
+						::Process.kill(:INT, @process_id)
+					end
+				end
+				
+				# Send `SIGTERM` to the child process.
+				def terminate!
+					unless @status
+						::Process.kill(:TERM, @process_id)
+					end
+				end
+				
+				# Send `SIGKILL` to the child process.
+				def kill!
+					unless @status
+						::Process.kill(:KILL, @process_id)
+					end
+				end
+				
+				# Send `SIGHUP` to the child process.
+				def restart!
+					unless @status
+						::Process.kill(:HUP, @process_id)
+					end
+				end
+				
+				# Reap the child process.
+				# @parameter timeout [Numeric | Nil] The maximum time to wait for the process to exit.
+				# @returns [::Process::Status | Nil] The process status, or `nil` if the timeout expired.
+				def reap(timeout = nil)
+					unless @status
+						if timeout
+							deadline = Deadline.new(timeout)
+							
+							loop do
+								if result = ::Process.waitpid2(@process_id, ::Process::WNOHANG)
+									_, @status = result
+									break
+								elsif deadline.expired?
+									return nil
+								else
+									sleep([deadline.remaining, 0.01].min)
+								end
+							end
+						else
+							_, @status = ::Process.waitpid2(@process_id)
+						end
+					end
+					
+					return @status
+				end
+			end
+		end
+	end
+end
diff --git a/lib/async/container/forked/instance.rb b/lib/async/container/forked/instance.rb
new file mode 100644
index 0000000..a724540
--- /dev/null
+++ b/lib/async/container/forked/instance.rb
@@ -0,0 +1,88 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require_relative "../notify/pipe"
+
+module Async
+	module Container
+		module Forked
+			# Represents a running child process from the point of view of the child process.
+			class Instance < Notify::Pipe
+				# Wrap an instance around the {Channel} instance from within the forked child.
+				# @parameter channel [Channel] The channel to use for communication.
+				# @parameter name [String] The name of the process.
+				def self.for(channel, name: nil)
+					instance = self.new(channel.out)
+					
+					# The child process won't be reading from the channel:
+					channel.close_read
+					
+					instance.name = name
+					
+					return instance
+				end
+				
+				# Initialize the child process instance.
+				#
+				# @parameter io [IO] The IO object to use for communication.
+				def initialize(io)
+					super
+					
+					@name = nil
+				end
+				
+				# Generate a hash representation of the process.
+				#
+				# @returns [Hash] The process as a hash, including `process_id` and `name`.
+				def as_json(...)
+					{
+						process_id: ::Process.pid,
+						name: @name,
+					}
+				end
+				
+				# Generate a JSON representation of the process.
+				#
+				# @returns [String] The process as JSON.
+				def to_json(...)
+					as_json.to_json(...)
+				end
+				
+				# Set the process title to the specified value.
+				#
+				# @parameter value [String] The name of the process.
+				def name= value
+					@name = value
+					
+					# This sets the process title to an empty string if the name is nil:
+					::Process.setproctitle(@name.to_s)
+				end
+				
+				# @returns [String] The name of the process.
+				def name
+					@name
+				end
+				
+				# Replace the current child process with a different one. Forwards arguments and options to {::Process.exec}.
+				# This method replaces the child process with the new executable, thus this method never returns.
+				#
+				# @parameter arguments [Array] The arguments to pass to the new process.
+				# @parameter ready [Boolean] If true, informs the parent process that the child is ready before exec. The notification pipe will still be passed to the exec'd process to prevent premature termination.
+				# @parameter options [Hash] Additional options to pass to {::Process.exec}.
+				def exec(*arguments, ready: true, **options)
+					# Always set up the notification pipe to be inherited by the exec'd process.
+					# This prevents the pipe from closing, which would trigger hang prevention and SIGKILL.
+					self.before_spawn(arguments, options)
+					
+					if ready
+						self.ready!(status: "(exec)")
+					end
+					
+					::Process.exec(*arguments, **options)
+				end
+			end
+		end
+	end
+end
diff --git a/lib/async/container/generic.rb b/lib/async/container/generic.rb
index 0364517..30851d8 100644
--- a/lib/async/container/generic.rb
+++ b/lib/async/container/generic.rb
@@ -2,10 +2,12 @@
 
 # Released under the MIT License.
 # Copyright, 2019-2026, by Samuel Williams.
-# Copyright, 2025, by Marc-André Cournoyer.
 
 require "etc"
 require "async/clock"
+require "async/deadline"
+require "set"
+require "thread"
 
 require_relative "group"
 require_relative "statistics"
@@ -16,9 +18,7 @@ module Container
 		# An environment variable key to override {.processor_count}.
 		ASYNC_CONTAINER_PROCESSOR_COUNT = "ASYNC_CONTAINER_PROCESSOR_COUNT"
 		
-		# The processor count which may be used for the default number of container threads/processes. You can override the value provided by the system by specifying the `ASYNC_CONTAINER_PROCESSOR_COUNT` environment variable.
-		# @returns [Integer] The number of hardware processors which can run threads/processes simultaneously.
-		# @raises [RuntimeError] If the process count is invalid.
+		# The processor count which may be used for the default number of container threads/processes.
 		def self.processor_count(env = ENV)
 			count = env.fetch(ASYNC_CONTAINER_PROCESSOR_COUNT) do
 				Etc.nprocessors rescue 1
@@ -31,281 +31,166 @@ def self.processor_count(env = ENV)
 			return count
 		end
 		
-		# A base class for implementing containers.
+		# A generic container which supervises children of a specific type.
 		class Generic
-			# Run a new container.
+			UNNAMED = "Unnamed"
+			
+			# Create and run a generic container.
+			# @parameter arguments [Array] Positional arguments for {#run}.
+			# @parameter options [Hash] Keyword options for {#run}.
+			# @returns [Generic] The running container.
 			def self.run(...)
 				self.new.run(...)
 			end
 			
-			UNNAMED = "Unnamed"
-			
-			# Initialize the container.
-			#
-			# @parameter policy [Policy] The policy to use for managing child lifecycle events.
-			# @parameter options [Hash] Options passed to the {Group} instance.
-			def initialize(policy: Policy::DEFAULT, **options)
+			# Initialize the generic container.
+			# @parameter child_type [Class] The child type used to spawn children.
+			# @parameter policy [Policy] The policy for managing child lifecycle events.
+			# @parameter options [Hash] Additional group options.
+			def initialize(child_type, policy: Policy::DEFAULT, **options)
+				@child_type = child_type
 				@group = Group.new(**options)
-				@stopping = false
-				
-				@state = {}
 				
 				@policy = policy
 				@statistics = @policy.make_statistics
+				
 				@keyed = {}
+				@threads = Set.new
+				@mutex = Mutex.new
+				@stopping = false
 			end
 			
-			# @attribute [Group] The group of running children instances.
+			# @attribute [Group] The group of running children.
 			attr :group
 			
-			# @returns [Integer] The number of running children instances.
-			def size
-				@group.size
-			end
-			
-			# @attribute [Hash(Child, Hash)] The state of each child instance.
-			attr :state
-			
 			# @attribute [Policy] The policy for managing child lifecycle events.
 			attr_accessor :policy
 			
-			# A human readable representation of the container.
-			# @returns [String]
-			def to_s
-				"#{self.class} with #{@statistics.spawns} spawns and #{@statistics.failures} failures."
+			# @attribute [Statistics] Statistics relating to child lifecycle.
+			attr :statistics
+			
+			# @returns [Integer] The number of running children.
+			def size
+				@group.size
 			end
 			
-			# Look up a child process by key.
-			# A key could be a symbol, a file path, or something else which the child instance represents.
-			def [] key
-				@keyed[key]
+			# @returns [Hash(Child, Hash)] The current state for each child.
+			def state
+				@group.children.each_with_object({}) do |child, state|
+					state[child] = child.state
+				end
 			end
 			
-			# Statistics relating to the behavior of children instances.
-			# @attribute [Statistics]
-			attr :statistics
+			# Look up a child by key.
+			def [](key)
+				@mutex.synchronize{@keyed[key]}
+			end
 			
 			# Whether any failures have occurred within the container.
-			# @returns [Boolean]
 			def failed?
 				@statistics.failed?
 			end
 			
-			# Whether the container has running children instances.
+			# Whether the container has running children.
 			def running?
 				@group.running?
 			end
 			
-			# Whether the container is currently stopping.
-			# @returns [Boolean]
+			# Whether the container is stopping.
 			def stopping?
-				@stopping
+				@mutex.synchronize{@stopping}
 			end
 			
-			# Sleep until some state change occurs or the specified duration elapses.
-			#
-			# @parameter duration [Numeric] the maximum amount of time to sleep for.
+			# Sleep until a group event occurs or the specified duration elapses.
 			def sleep(duration = nil)
-				@group.sleep(duration)
+				deadline = Deadline.new(duration) if duration
+				
+				loop do
+					event = if deadline
+						return nil if deadline.expired?
+						
+						@group.wait(deadline.remaining)
+					else
+						@group.wait
+					end
+					
+					case event
+					when Group::Spawn
+						next
+					else
+						return event
+					end
+				end
 			end
 			
-			# Wait until all spawned tasks are completed.
+			# Wait until all lifecycle owner threads complete.
 			def wait
-				@group.wait
+				loop do
+					current = Thread.current
+					threads = @mutex.synchronize{@threads.reject{|thread| thread.equal?(current)}}
+					break if threads.empty?
+					
+					threads.each do |thread|
+						thread.join
+						@mutex.synchronize{@threads.delete(thread)}
+					end
+				end
 			end
 			
-			# Gracefully interrupt all child instances.
+			# Interrupt all children and enter the stopping state.
 			def interrupt
-				# We must enter the stopping state before signalling the children. Interrupting a child causes it to drain and exit, but the main run loop will respawn any child that exits while `restart: true` and the container is not stopping (see the `restart && !@stopping` gate in `#run`). Without setting this flag, an interrupted child immediately respawns, so the container never drains and `#wait` never returns.
-				#
-				# This matters most for `Hybrid` containers: a `SIGINT`/`SIGTERM` delivered to a fork is translated into a call to `#interrupt` on the inner threaded container, which typically runs with `restart: true` (the default for `async-service` managed services). If `#interrupt` did not set this flag, the inner threads would drain, exit, and respawn in a loop, so a single signal would never terminate the fork. Setting `@stopping = true` here makes `#interrupt` behave as the start of a graceful shutdown: children drain and exit, are not respawned, and the fork terminates - consistent with how `Forked` and `Threaded` containers handle a single interrupt.
-				@stopping = true
-				@group.interrupt
+				stopping!
+				@group.interrupt!
 			end
 			
-			# Returns true if all children instances have the specified status flag set.
-			# e.g. `:ready`.
-			# This state is updated by the process readiness protocol mechanism. See {Notify::Client} for more details.
-			# @returns [Boolean]
+			# Returns true if all running children have the specified status flag.
 			def status?(flag)
-				# This also returns true if all processes have exited/failed:
-				@state.all?{|_, state| state[flag]}
+				@group.children.all?{|child| child.state[flag]}
 			end
 			
-			# Wait until all the children instances have indicated that they are ready.
-			# @returns [Boolean] The children all became ready.
+			# Wait until all running children report readiness.
 			def wait_until_ready
-				while true
-					Console.debug(self) do |buffer|
-						buffer.puts "Waiting for ready:"
-						@state.each do |child, state|
-							buffer.puts "\t#{child.inspect}: #{state}"
-						end
-					end
+				loop do
+					return true if running? && status?(:ready)
+					return false if failed? && !lifecycle_running?
+					return true if !running? && !lifecycle_running?
 					
-					self.sleep
-					
-					if self.status?(:ready)
-						Console.debug(self) do |buffer|
-							buffer.puts "All ready:"
-							@state.each do |child, state|
-								buffer.puts "\t#{child.inspect}: #{state}"
-							end
-						end
-						
-						return true
-					end
+					@group.wait(0.1)
 				end
 			end
 			
-			# Stop the children instances.
-			# @parameter timeout [Boolean | Numeric] Whether to stop gracefully, or a specific timeout.
+			# Stop all children.
 			def stop(timeout = true)
-				if @stopping
-					Console.warn(self, "Container is already stopping!")
-					return
-				end
+				return if stopping? && !running?
 				
-				Console.info(self, "Stopping container...", timeout: timeout)
-				@stopping = true
+				stopping!
 				@group.stop(timeout)
-				
-				if @group.running?
-					Console.warn(self, "Group is still running after stopping it!")
-				else
-					Console.info(self, "Group has stopped.")
-				end
-			rescue => error
-				Console.error(self, "Error while stopping container!", exception: error)
-				raise
-			end
-			
-			protected def health_check_failed(child, age_clock, health_check_timeout)
-				begin
-					@policy.health_check_failed(
-						self, child,
-						age: age_clock.total,
-						timeout: health_check_timeout
-					)
-				rescue => error
-					Console.error(self, "Policy error in health_check_failed!", exception: error)
-					child.kill!
-				end
-			end
-			
-			protected def startup_failed(child, age_clock, startup_timeout)
-				begin
-					@policy.startup_failed(
-						self, child,
-						age: age_clock.total,
-						timeout: startup_timeout
-					)
-				rescue => error
-					Console.error(self, "Policy error in startup_failed!", exception: error)
-					child.kill!
-				end
+				self.wait
 			end
 			
-			# Spawn a child instance into the container.
-			# @parameter name [String] The name of the child instance.
-			# @parameter restart [Boolean] Whether to restart the child instance if it fails.
-			# @parameter key [Symbol] An optional key used to look up (via {[]}) and reuse the child instance.
-			# @parameter health_check_timeout [Numeric | Nil] The maximum time a child instance can run without updating its state, before it is terminated as unhealthy.
-			# @parameter startup_timeout [Numeric | Nil] The maximum time a child instance can run without becoming ready, before it is terminated as unhealthy.
-			def spawn(name: nil, restart: false, key: nil, health_check_timeout: nil, startup_timeout: nil, &block)
+			# Spawn a child into the container.
+			def spawn(name: nil, restart: false, key: nil, health_check_timeout: nil, startup_timeout: nil, **options, &block)
 				name ||= UNNAMED
 				
 				if reuse?(key)
-					Console.debug(self, "Reusing existing child.", child: {key: key, name: name})
 					return false
 				end
 				
-				@statistics.spawn!
+				child = start_child(name: name, **options, &block)
+				register_child(child, name: name, key: key)
 				
-				fiber do
-					until @stopping
-						Console.debug(self, "Starting child...", child: {key: key, name: name, restart: restart, health_check_timeout: health_check_timeout}, statistics: @statistics)
-						
-						child = self.start(name, &block)
-						state = insert(key, child)
-						
-						# Notify policy of spawn
-						begin
-							@policy.child_spawn(self, child, name: name, key: key)
-						rescue => error
-							Console.error(self, "Policy error in child_spawn!", exception: error)
-						end
-						
-						Console.debug(self, "Started child.", child: child, spawn: {key: key, restart: restart, health_check_timeout: health_check_timeout}, statistics: @statistics)
-						
-						# If a health check or startup timeout is specified, we will monitor the child process and terminate it if it does not update its state within the specified time.
-						if health_check_timeout || startup_timeout
-							age_clock = state[:age] = Clock.start
-						end
-						
-						status = nil
-						
-						begin
-							status = @group.wait_for(child) do |message|
-								case message
-								when :health_check!
-									if state[:ready]
-										# If a health check timeout is specified, we will monitor the child process and terminate it if it does not update its state within the specified time.
-										if health_check_timeout
-											if health_check_timeout < age_clock.total
-												health_check_failed(child, age_clock, health_check_timeout)
-											end
-										end
-									else
-										# If a startup timeout is specified, we will monitor the child process and terminate it if it does not become ready within the specified time.
-										if startup_timeout
-											if startup_timeout < age_clock.total
-												startup_failed(child, age_clock, startup_timeout)
-											end
-										end
-									end
-								else
-									state.update(message)
-									
-									# Reset the age clock if the child has become ready:
-									if state[:ready]
-										age_clock&.reset!
-									end
-								end
-							end
-						rescue => error
-							Console.error(self, "Error during child process management!", exception: error, stopping: @stopping)
-						ensure
-							delete(key, child)
-						end
-						
-						if status&.success?
-							Console.debug(self, "Child exited successfully.", status: status, stopping: @stopping)
-						else
-							@statistics.failure!
-							Console.error(self, "Child exited with error!", status: status, stopping: @stopping)
-						end
-						
-						# Notify policy of exit (after statistics are updated):
-						begin
-							@policy.child_exit(self, child, status, name: name, key: key)
-						rescue => error
-							Console.error(self, "Policy error in child_exit!", exception: error)
-						end
-						
-						if restart && !@stopping
-							@statistics.restart!
-						else
-							break
-						end
-					end
-				end.resume
+				thread = Thread.new do
+					manage_child(child, name: name, key: key, restart: restart, health_check_timeout: health_check_timeout, startup_timeout: startup_timeout, options: options, block: block)
+				ensure
+					@mutex.synchronize{@threads.delete(Thread.current)}
+				end
+				
+				@mutex.synchronize{@threads.add(thread)}
 				
 				return true
 			end
 			
 			# Run multiple instances of the same block in the container.
-			# @parameter count [Integer] The number of instances to start.
 			def run(count: Container.processor_count, **options, &block)
 				count.times do
 					spawn(**options, &block)
@@ -316,8 +201,6 @@ def run(count: Container.processor_count, **options, &block)
 			
 			# @deprecated Please use {spawn} or {run} instead.
 			def async(**options, &block)
-				# warn "#{self.class}##{__method__} is deprecated, please use `spawn` or `run` instead.", uplevel: 1
-				
 				require "async"
 				
 				spawn(**options) do |instance|
@@ -326,65 +209,167 @@ def async(**options, &block)
 			end
 			
 			# Re-run the given block against the container.
-			#
-			# Existing keyed children are reused (see {spawn}), so re-running setup will not
-			# duplicate them. Reconciliation of children whose keys are no longer configured
-			# (i.e. stopping obsolete children) is not currently supported and will be revisited.
 			def reload
 				yield
 			end
 			
-			# Whether a child instance already exists for the given key, in which case it can be reused rather than spawned again.
-			def reuse?(key)
-				if key
-					@keyed.key?(key)
-				else
-					false
-				end
+			# Whether a child exists for the given key.
+			def key?(key)
+				key && @mutex.synchronize{@keyed.key?(key)}
 			end
 			
-			# Whether a child instance exists for the given key.
-			def key?(key)
-				if key
-					@keyed.key?(key)
-				end
+			# Whether a child can be reused for the given key.
+			def reuse?(key)
+				key && @mutex.synchronize{@keyed.key?(key)}
 			end
 			
 			protected
 			
-			# Register the child (value) as running.
-			def insert(key, child)
-				if key
-					@keyed[key] = child
-				end
+			def lifecycle_running?
+				@mutex.synchronize{@threads.any?(&:alive?)}
+			end
+			
+			def stopping!
+				@mutex.synchronize{@stopping = true}
+			end
+			
+			def start_child(name:, **options, &block)
+				@child_type.call(name: name, **options, &block)
+			end
+			
+			def register_child(child, name:, key:)
+				@statistics.spawn!
 				
-				state = {}
+				@mutex.synchronize do
+					@keyed[key] = child if key
+				end
 				
-				@state[child] = state
+				@group.add(child)
 				
-				return state
+				begin
+					@policy.child_spawn(self, child, name: name, key: key)
+				rescue => error
+					Console.error(self, "Policy error in child_spawn!", exception: error) if defined?(Console)
+				end
 			end
 			
-			# Clear the child (value) as running.
-			def delete(key, child)
-				if key
-					@keyed.delete(key)
+			def unregister_child(child, status, key:)
+				@mutex.synchronize do
+					@keyed.delete(key) if key
 				end
 				
-				@state.delete(child)
+				@group.remove(child, status)
 			end
 			
-			private
+			def manage_child(child, name:, key:, restart:, health_check_timeout:, startup_timeout:, options:, block:)
+				loop do
+					status = monitor_child(child, health_check_timeout: health_check_timeout, startup_timeout: startup_timeout)
+					
+					record_exit(child, status, name: name, key: key)
+					unregister_child(child, status, key: key)
+					notify_child_exit(child, status, name: name, key: key)
+					
+					if restart && !stopping?
+						@statistics.restart!
+						
+						child = start_child(name: name, **options, &block)
+						register_child(child, name: name, key: key)
+					else
+						break
+					end
+				end
+			rescue => error
+				Console.error(self, "Error during child lifecycle management!", exception: error, stopping: stopping?) if defined?(Console)
+			ensure
+				if @group.children.include?(child)
+					begin
+						@group.remove(child)
+					rescue ArgumentError
+					end
+				end
+			end
+			
+			def monitor_child(child, health_check_timeout:, startup_timeout:)
+				if health_check_timeout || startup_timeout
+					monitor_child_with_timeouts(child, health_check_timeout: health_check_timeout, startup_timeout: startup_timeout)
+				else
+					child.wait do |message|
+						@group.update(child, message)
+					end
+				end
+			end
 			
-			if Fiber.respond_to?(:blocking?)
-				def fiber(&block)
-					Fiber.new(blocking: true, &block)
+			def monitor_child_with_timeouts(child, health_check_timeout:, startup_timeout:)
+				startup_deadline = Deadline.new(startup_timeout) if startup_timeout
+				startup_started_at = Clock.now if startup_timeout
+				
+				loop do
+					timeout = if child.ready?
+						health_check_timeout
+					elsif startup_deadline
+						if startup_deadline.expired?
+							false
+						else
+							startup_deadline.remaining
+						end
+					end
+					
+					result = timeout == false ? false : child.receive(timeout)
+					
+					case result
+					when false
+						if child.ready?
+							health_check_failed(child, health_check_timeout)
+						else
+							startup_failed(child, startup_timeout, age: Clock.now - startup_started_at)
+						end
+						
+						child.kill!
+						
+						return child.wait do |message|
+							@group.update(child, message)
+						end
+					when nil
+						return child.reap
+					else
+						@group.update(child, result)
+					end
 				end
-			else
-				def fiber(&block)
-					Fiber.new(&block)
+			end
+			
+			def record_exit(child, status, name:, key:)
+				stopping = stopping?
+				
+				if status&.success?
+					Console.debug(self, "Child exited successfully.", status: status, stopping: stopping) if defined?(Console)
+				elsif stopping
+					Console.debug(self, "Child exited while stopping.", status: status, stopping: stopping) if defined?(Console)
+				else
+					@statistics.failure!
+					Console.error(self, "Child exited with error!", status: status, stopping: stopping) if defined?(Console)
 				end
 			end
+			
+			def notify_child_exit(child, status, name:, key:)
+				@policy.child_exit(self, child, status, name: name, key: key)
+			rescue => error
+				Console.error(self, "Policy error in child_exit!", exception: error) if defined?(Console)
+				
+			end
+			
+			def health_check_failed(child, timeout)
+				@policy.health_check_failed(self, child, age: Clock.now - child.last_updated_at, timeout: timeout)
+			rescue => error
+				Console.error(self, "Policy error in health_check_failed!", exception: error) if defined?(Console)
+				child.kill!
+			end
+			
+			def startup_failed(child, timeout, age: Clock.now - child.last_updated_at)
+				@policy.startup_failed(self, child, age: age, timeout: timeout)
+			rescue => error
+				Console.error(self, "Policy error in startup_failed!", exception: error) if defined?(Console)
+				child.kill!
+			end
 		end
 	end
 end
diff --git a/lib/async/container/group.rb b/lib/async/container/group.rb
index def3182..e68d356 100644
--- a/lib/async/container/group.rb
+++ b/lib/async/container/group.rb
@@ -1,239 +1,167 @@
 # frozen_string_literal: true
 
 # Released under the MIT License.
-# Copyright, 2018-2026, by Samuel Williams.
+# Copyright, 2026, by Samuel Williams.
 
-require "fiber"
-require "async/clock"
-
-require_relative "error"
+require "async/deadline"
+require "set"
 
 module Async
 	module Container
-		# The default timeout for graceful termination, used when the `graceful` argument is true.
-		DEFAULT_GRACEFUL_TIMEOUT = 10.0
-		
-		# Manages a group of running processes.
+		# Tracks a set of children and publishes state-change events.
 		class Group
-			# Initialize an empty group.
-			#
-			# @parameter health_check_interval [Numeric | Nil] The (biggest) interval at which health checks are performed.
-			def initialize(health_check_interval: 1.0)
-				@health_check_interval = health_check_interval
-				
-				# The running fibers, indexed by IO:
-				@running = {}
-			end
+			# Emitted when a child is spawned into the group.
+			Spawn = Data.define(:child)
 			
-			# @returns [String] A human-readable representation of the group.
-			def inspect
-				"#<#{self.class} running=#{@running.size}>"
-			end
-			
-			# @attribute [Hash(IO, Fiber)] the running tasks, indexed by IO.
-			attr :running
+			# Emitted when a child sends a notification message.
+			Update = Data.define(:child, :message)
 			
-			# @returns [Integer] The number of running processes.
-			def size
-				@running.size
-			end
+			# Emitted when a child exits the group.
+			Exit = Data.define(:child, :status)
 			
-			# Whether the group contains any running processes.
-			# @returns [Boolean]
-			def running?
-				@running.any?
+			# Initialize the group.
+			def initialize
+				@children = Set.new
+				@events = Thread::Queue.new
 			end
 			
-			# Whether the group contains any running processes.
-			# @returns [Boolean]
-			def any?
-				@running.any?
-			end
+			# @attribute [Set(Child)] The children currently in the group.
+			attr :children
 			
-			# Whether the group is empty.
-			# @returns [Boolean]
-			def empty?
-				@running.empty?
+			# @returns [Set(Child)] The children currently running.
+			def running
+				@children
 			end
 			
-			# Sleep for at most the specified duration until some state change occurs.
-			def sleep(duration)
-				self.wait_for_children(duration)
+			# @returns [Integer] The number of children in the group.
+			def size
+				@children.size
 			end
 			
-			# Begin any outstanding queued processes and wait for them indefinitely.
-			def wait
-				with_health_checks do |duration|
-					self.wait_for_children(duration)
-				end
+			# @returns [Boolean] Whether the group has any children.
+			def any?
+				@children.any?
 			end
 			
-			private def with_health_checks
-				if @health_check_interval
-					health_check_clock = Clock.start
-					
-					while self.running?
-						duration = [@health_check_interval - health_check_clock.total, 0].max
-						
-						yield duration
-						
-						if health_check_clock.total > @health_check_interval
-							self.health_check!
-							health_check_clock.reset!
-						end
-					end
-				else
-					while self.running?
-						yield nil
-					end
-				end
+			# @returns [Boolean] Whether the group has no children.
+			def empty?
+				@children.empty?
 			end
 			
-			private def each_running(&block)
-				# We create a copy of the values here, in case the block modifies the running set:
-				@running.values.each(&block)
+			# @returns [Boolean] Whether the group has any children.
+			def running?
+				any?
 			end
 			
-			# Perform a health check on all running processes.
-			def health_check!
-				each_running do |fiber|
-					fiber.resume(:health_check!)
-				end
+			# Iterate over the children in the group.
+			def each(&block)
+				@children.each(&block)
 			end
 			
-			# Interrupt all running processes.
-			# This resumes the controlling fiber with an instance of {Interrupt}.
-			def interrupt
-				Console.info(self, "Sending interrupt to #{@running.size} running processes...")
-				each_running do |fiber|
-					fiber.resume(Interrupt)
+			# Add a child to the group.
+			def add(child)
+				unless @children.add?(child)
+					raise ArgumentError, "Child already exists in group: #{child.inspect}"
 				end
+				
+				@events.push(Spawn.new(child))
+				
+				return child
 			end
 			
-			# Terminate all running processes.
-			# This resumes the controlling fiber with an instance of {Terminate}.
-			def terminate
-				Console.info(self, "Sending terminate to #{@running.size} running processes...")
-				each_running do |fiber|
-					fiber.resume(Terminate)
-				end
+			# Publish a child notification event.
+			def update(child, message)
+				@events.push(Update.new(child, message))
+				
+				return child
 			end
 			
-			# Kill all running processes.
-			# This resumes the controlling fiber with an instance of {Kill}.
-			def kill
-				Console.info(self, "Sending kill to #{@running.size} running processes...")
-				each_running do |fiber|
-					fiber.resume(Kill)
-				end
-			end
-			
-			private def wait_for_exit(clock, timeout)
-				while self.any?
-					duration = timeout - clock.total
-					
-					if duration >= 0
-						self.wait_for_children(duration)
-					else
-						self.wait_for_children(0)
-						break
-					end
+			# Remove a child from the group.
+			def remove(child, status = nil)
+				unless @children.delete?(child)
+					raise ArgumentError, "Child does not exist in group: #{child.inspect}"
 				end
+				
+				@events.push(Exit.new(child, status))
+				
+				return child
 			end
 			
-			# Stop all child processes with a multi-phase shutdown sequence.
+			# Wait for group events.
 			#
-			# A graceful shutdown performs the following sequence:
-			# 1. Send SIGINT and wait up to `graceful` seconds if specified.
-			# 2. Send SIGKILL and wait indefinitely for process cleanup.
-			#
-			# If `graceful` is true, default to `DEFAULT_GRACEFUL_TIMEOUT` (10 seconds).
-			# If `graceful` is false, skip the SIGINT phase and go directly to SIGKILL.
-			#
-			# @parameter graceful [Boolean | Numeric] Whether to send SIGINT first or skip directly to SIGKILL.
-			def stop(graceful = true)
-				Console.debug(self, "Stopping all processes...", graceful: graceful)
+			# If a block is given, events are yielded until the block breaks, the
+			# timeout expires, or the event queue closes. If no block is given, a
+			# single event is returned.
+			def wait(timeout = nil)
+				deadline = Deadline.new(timeout) if timeout
 				
-				# If a timeout is specified, interrupt the children first:
-				if graceful
-					# Send SIGINT to the children:
-					self.interrupt
-					
-					if graceful == true
-						graceful = DEFAULT_GRACEFUL_TIMEOUT
+				loop do
+					event = if deadline
+						return nil if deadline.expired?
+						
+						@events.pop(timeout: deadline.remaining)
+					else
+						@events.pop
 					end
 					
-					clock = Clock.start
-					
-					# Wait for the children to exit:
-					self.wait_for_exit(clock, graceful)
-				end
-			ensure
-				# Do our best to clean up the children:
-				if any?
-					if graceful
-						Console.warn(self, "Killing processes after graceful shutdown failed...", size: self.size, clock: clock)
-					end
+					return nil unless event
+					return event unless block_given?
 					
-					self.kill
-					self.wait
+					yield event
 				end
 			end
 			
-			# Wait for a message in the specified {Channel}.
-			def wait_for(channel)
-				io = channel.in
-				
-				@running[io] = Fiber.current
+			# Wait until the group has no children.
+			def wait_until_empty(timeout = nil, &block)
+				return true if empty?
 				
-				while @running.key?(io)
-					# Wait for some event on the channel:
-					result = Fiber.yield
-					
-					if result == Interrupt
-						channel.interrupt!
-					elsif result == Terminate
-						channel.terminate!
-					elsif result == Kill
-						channel.kill!
-					elsif result
-						yield result
-					elsif message = channel.receive
-						yield message
-					else
-						# Wait for the channel to exit:
-						return channel.wait
-					end
+				result = wait(timeout) do |event|
+					yield event if block
+					break true if empty?
 				end
-			ensure
-				@running.delete(io)
+				
+				return result == true
 			end
 			
-			protected
+			# Interrupt all children in the group.
+			def interrupt!
+				@children.each(&:interrupt!)
+			end
 			
-			def wait_for_children(duration = nil)
-				# This log is a bit noisy and doesn't really provide a lot of useful information:
-				Console.debug(self, "Waiting for children...", duration: duration, running: @running)
-				
-				unless @running.empty?
-					# Maybe consider using a proper event loop here:
-					if ready = self.select(duration)
-						ready.each do |io|
-							if fiber = @running[io]
-								# This method can be re-entered. While resuming a fiber, a policy hook may be invoked, which may invoke operations on the container. In that case, select may be called again on the same set of waiting fibers. On returning, those fibers may have already completed and removed themselves from @running, so we need to check for that.
-								fiber.resume
-							end
-						end
-					end
-				end
+			# Terminate all children in the group.
+			def terminate!
+				@children.each(&:terminate!)
+			end
+			
+			# Kill all children in the group.
+			def kill!
+				@children.each(&:kill!)
 			end
 			
-			# Wait for a child process to exit OR a signal to be received.
-			def select(duration)
-				::Thread.handle_interrupt(SignalException => :immediate) do
-					readable, _, _ = ::IO.select(@running.keys, nil, nil, duration)
+			# Restart all children in the group.
+			def restart!
+				@children.each(&:restart!)
+			end
+			
+			# Stop all children in the group without consuming child channels.
+			#
+			# A graceful shutdown sends interrupts and waits for child owner tasks to
+			# remove the children. If a numeric timeout expires, remaining children are
+			# killed and the method waits indefinitely for removal. If `graceful` is
+			# false, children are killed immediately.
+			def stop(graceful = true, &block)
+				return true if empty?
+				
+				if graceful
+					interrupt!
 					
-					return readable
+					timeout = (graceful == true) ? nil : graceful
+					return true if wait_until_empty(timeout, &block)
+				end
+				
+				unless empty?
+					kill!
+					wait_until_empty(nil, &block)
 				end
 			end
 		end
diff --git a/lib/async/container/hybrid.rb b/lib/async/container/hybrid.rb
index 8c6f4f5..44eee61 100644
--- a/lib/async/container/hybrid.rb
+++ b/lib/async/container/hybrid.rb
@@ -1,50 +1,65 @@
 # frozen_string_literal: true
 
 # Released under the MIT License.
-# Copyright, 2019-2025, by Samuel Williams.
-# Copyright, 2022, by Anton Sozontov.
+# Copyright, 2019-2026, by Samuel Williams.
 
 require_relative "forked"
 require_relative "threaded"
 
 module Async
 	module Container
-		# Provides a hybrid multi-process multi-thread container.
-		class Hybrid < Forked
-			# Run multiple instances of the same block in the container.
-			# @parameter count [Integer] The number of instances to start.
-			# @parameter forks [Integer] The number of processes to fork.
-			# @parameter threads [Integer] the number of threads to start.
-			# @parameter health_check_timeout [Numeric] The timeout for health checks, in seconds. Passed into the child {Threaded} containers.
+		# Represents a container which spawns forked processes that manage threaded children.
+		class Hybrid < Generic
+			# Initialize the hybrid container.
+			# @parameter arguments [Array] Positional arguments for {Generic#initialize}.
+			# @parameter options [Hash] Keyword options for {Generic#initialize}.
+			def initialize(*arguments, **options)
+				super(Forked::Child, *arguments, **options)
+			end
+			
+			# Spawn forked containers, each managing a set of threaded children.
+			# @parameter count [Integer | Nil] The total number of threaded children.
+			# @parameter forks [Integer | Nil] The number of forked child containers.
+			# @parameter threads [Integer | Nil] The number of threads per fork.
+			# @parameter health_check_timeout [Numeric | Nil] The timeout for child health checks.
+			# @parameter options [Hash] Additional child options.
+			# @yields {|instance| ...} The threaded child body.
+			# 	@parameter instance [Threaded::Instance] The child-side instance interface.
+			# @returns [Hybrid] The hybrid container.
 			def run(count: nil, forks: nil, threads: nil, health_check_timeout: nil, **options, &block)
-				processor_count = Container.processor_count
+				processor_count = Async::Container.processor_count
 				count ||= processor_count ** 2
 				forks ||= [processor_count, count].min
 				threads ||= (count / forks).ceil
 				
 				forks.times do
-					self.spawn(**options) do |instance|
+					spawn(**options) do |instance|
 						container = Threaded.new
 						
 						container.run(count: threads, health_check_timeout: health_check_timeout, **options, &block)
-						
 						container.wait_until_ready
+						
 						instance.ready!
 						
 						begin
 							container.wait
 						rescue Interrupt
-							# Gracefully interrupt child threads; parent process handles escalation.
 							container.interrupt
 							retry
 						end
 					ensure
-						container.stop(false)
+						container&.stop(false)
 					end
 				end
 				
 				return self
 			end
+			
+			# Whether this container uses multiple processes.
+			# @returns [Boolean]
+			def self.multiprocess?
+				true
+			end
 		end
 	end
 end
diff --git a/lib/async/container/threaded.rb b/lib/async/container/threaded.rb
index 5224519..247d001 100644
--- a/lib/async/container/threaded.rb
+++ b/lib/async/container/threaded.rb
@@ -4,290 +4,24 @@
 # Copyright, 2017-2026, by Samuel Williams.
 
 require_relative "generic"
-require_relative "channel"
-require_relative "notify/pipe"
+require_relative "threaded/child"
 
 module Async
 	module Container
-		# A multi-thread container which uses {Thread.fork}.
-		class Threaded < Generic
-			# Indicates that this is not a multi-process container.
-			def self.multiprocess?
-				false
-			end
-			
-			# Represents a running child thread from the point of view of the parent container.
-			class Child < Channel
-				# Used to propagate the exit status of a child process invoked by {Instance#exec}.
-				class Exit < Exception
-					# Initialize the exit status.
-					# @parameter status [::Process::Status] The process exit status.
-					def initialize(status)
-						@status = status
-					end
-					
-					# The process exit status.
-					# @attribute [::Process::Status]
-					attr :status
-					
-					# The process exit status if it was an error.
-					# @returns [::Process::Status | Nil]
-					def error
-						unless status.success?
-							status
-						end
-					end
-				end
-				
-				# Represents a running child thread from the point of view of the child thread.
-				class Instance < Notify::Pipe
-					# Wrap an instance around the {Thread} instance from within the threaded child.
-					# @parameter thread [Thread] The thread intance to wrap.
-					def self.for(thread)
-						instance = self.new(thread.out)
-						
-						return instance
-					end
-					
-					# Initialize the child thread instance.
-					#
-					# @parameter io [IO] The IO object to use for communication with the parent.
-					def initialize(io)
-						@thread = ::Thread.current
-						
-						super
-					end
-					
-					# Generate a hash representation of the thread.
-					#
-					# @returns [Hash] The thread as a hash, including `process_id`, `thread_id`, and `name`.
-					def as_json(...)
-						{
-							process_id: ::Process.pid,
-							thread_id: @thread.object_id,
-							name: @thread.name,
-						}
-					end
-					
-					# Generate a JSON representation of the thread.
-					#
-					# @returns [String] The thread as JSON.
-					def to_json(...)
-						as_json.to_json(...)
-					end
-					
-					# Set the name of the thread.
-					# @parameter value [String] The name to set.
-					def name= value
-						@thread.name = value
-					end
-					
-					# Get the name of the thread.
-					# @returns [String]
-					def name
-						@thread.name
-					end
-					
-					# Execute a child process using {::Process.spawn}. In order to simulate {::Process.exec}, an {Exit} instance is raised to propagage exit status.
-					# This creates the illusion that this method does not return (normally).
-					def exec(*arguments, ready: true, **options)
-						# Always set up the notification pipe to be inherited by the spawned process.
-						self.before_spawn(arguments, options)
-						
-						if ready
-							self.ready!(status: "(spawn)")
-						end
-						
-						begin
-							pid = ::Process.spawn(*arguments, **options)
-						ensure
-							_, status = ::Process.wait2(pid)
-							
-							raise Exit, status
-						end
-					end
-				end
-				
-				# Start a new child thread and execute the provided block in it.
-				#
-				# @parameter options [Hash] Additional options to to the new child instance.
-				def self.fork(**options)
-					self.new(**options) do |thread|
-						::Thread.new do
-							# Reset `SignalException` delivery because CRuby inherits the `Thread.handle_interrupt` mask stack across `Thread.new`. Async deliberately masks signal exceptions, and signal-driven thread control should be delivered promptly:
-							::Thread.handle_interrupt(SignalException => :immediate) do
-								yield Instance.for(thread)
-							end
-						end
-					end
-				end
-				
-				# Initialize the thread.
-				#
-				# @parameter name [String] The name to use for the child thread.
-				def initialize(name: nil, **options)
-					super(**options)
-					
-					@status = nil
-					
-					@thread = yield(self)
-					@thread.report_on_exception = false
-					@thread.name = name
-					
-					@waiter = ::Thread.new do
-						begin
-							@thread.join
-						rescue Exit => exit
-							finished(exit.error)
-						rescue Interrupt
-							# Graceful shutdown.
-							finished
-						rescue Exception => error
-							finished(error)
-						else
-							finished
-						end
-					end
-				end
-				
-				# Convert the child process to a hash, suitable for serialization.
-				#
-				# @returns [Hash] The request as a hash.
-				def as_json(...)
-					{
-						name: @thread.name,
-						status: @status&.as_json,
-					}
-				end
-				
-				# Convert the request to JSON.
-				#
-				# @returns [String] The request as JSON.
-				def to_json(...)
-					as_json.to_json(...)
-				end
-				
-				# Set the name of the thread.
-				# @parameter value [String] The name to set.
-				def name= value
-					@thread.name = value
-				end
-				
-				# Get the name of the thread.
-				# @returns [String]
-				def name
-					@thread.name
-				end
-				
-				# A human readable representation of the thread.
-				# @returns [String]
-				def to_s
-					"\#<#{self.class} #{@thread.name}>"
-				end
-				
-				# Invoke {#terminate!} and then {#wait} for the child thread to exit.
-				def close
-					self.terminate!
-					self.wait
-				ensure
-					super
-				end
-				
-				# Raise {Interrupt} in the child thread.
-				def interrupt!
-					@thread.raise(Interrupt)
-				end
-				
-				# Raise {Terminate} in the child thread.
-				def terminate!
-					@thread.raise(Terminate)
-				end
-				
-				# Invoke {Thread#kill} on the child thread.
-				def kill!
-					# Killing a thread does not raise an exception in the thread, so we need to handle the status here:
-					@status = Status.new(:killed)
-					
-					@thread.kill
-				end
-				
-				# Raise {Restart} in the child thread.
-				def restart!
-					@thread.raise(Restart)
-				end
-				
-				# Wait for the thread to exit and return he exit status.
-				# @asynchronous This method may block.
-				#
-				# @parameter timeout [Numeric | Nil] Maximum time to wait before forceful termination.
-				# @returns [Status]
-				def wait(timeout = 0.1)
-					if @waiter
-						Console.debug(self, "Waiting for thread to exit...", child: {thread_id: @thread.object_id}, timeout: timeout)
-						
-						unless @waiter.join(timeout)
-							Console.warn(self, "Thread is blocking, sending kill signal...", child: {thread_id: @thread.object_id}, timeout: timeout)
-							self.kill!
-							@waiter.join
-						end
-						
-						@waiter = nil
-					end
-					
-					Console.debug(self, "Thread exited.", child: {thread_id: @thread.object_id, status: @status})
-					
-					return @status
-				end
-				
-				# A pseudo exit-status wrapper.
-				class Status
-					# Initialise the status.
-					# @parameter error [::Process::Status] The exit status of the child thread.
-					def initialize(error = nil)
-						@error = error
-					end
-					
-					# Whether the status represents a successful outcome.
-					# @returns [Boolean]
-					def success?
-						@error.nil?
-					end
-					
-					# Convert the status to a hash, suitable for serialization.
-					#
-					# @returns [Boolean | String] If the status is an error, the error message is returned, otherwise `true`.
-					def as_json(...)
-						if @error
-							@error.inspect
-						else
-							true
-						end
-					end
-					
-					# A human readable representation of the status.
-					def to_s
-						"\#<#{self.class} #{success? ? "success" : "failure"}>"
-					end
-				end
-				
-				protected
-				
-				# Invoked by the @waiter thread to indicate the outcome of the child thread.
-				def finished(error = nil)
-					if error
-						Console.error(self){error}
-					end
-					
-					@status ||= Status.new(error)
-					self.close_write
-				end
+		# Public factory for multi-thread containers.
+		module Threaded
+			# Create a generic container which spawns threaded children.
+			# @parameter arguments [Array] Positional arguments for {Generic#initialize}.
+			# @parameter options [Hash] Keyword options for {Generic#initialize}.
+			# @returns [Generic] A generic container configured with {Threaded::Child}.
+			def self.new(*arguments, **options)
+				Generic.new(Child, *arguments, **options)
 			end
 			
-			# Start a named child thread and execute the provided block in it.
-			# @parameter name [String] The name (title) of the child process.
-			# @parameter block [Proc] The block to execute in the child process.
-			def start(name, &block)
-				Child.fork(name: name, &block)
+			# Whether this container uses multiple processes.
+			# @returns [Boolean]
+			def self.multiprocess?
+				false
 			end
 		end
 	end
diff --git a/lib/async/container/threaded/child.rb b/lib/async/container/threaded/child.rb
new file mode 100644
index 0000000..ede1104
--- /dev/null
+++ b/lib/async/container/threaded/child.rb
@@ -0,0 +1,130 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require_relative "../child"
+require_relative "instance"
+
+module Async
+	module Container
+		module Threaded
+			# Represents a child thread managed by a threaded container.
+			class Child < Container::Child
+				# Start a child thread.
+				# @parameter channel [Channel] The notification channel for the child.
+				# @parameter name [String | Nil] The optional child name.
+				# @parameter options [Hash] Additional child options.
+				# @yields {|instance| ...} The child thread body.
+				# 	@parameter instance [Instance] The child-side instance interface.
+				# @returns [Child] The threaded child.
+				def self.call(channel: Channel.new, name: nil, **options, &block)
+					thread = ::Thread.new do
+						begin
+							yield Instance.for(channel, name: name)
+						rescue Exit => exit
+							Status.new(exit.error)
+						rescue Interrupt
+							# Graceful shutdown.
+							Status.new
+						rescue Exception => error
+							Status.new(error)
+						else
+							Status.new
+						ensure
+							# Closing the write end tells the parent that no more messages are coming:
+							channel.close_write unless channel.out.closed?
+						end
+					end
+					
+					return self.new(thread, channel, name: name, **options)
+				end
+				
+				# Initialize the child thread wrapper.
+				# @parameter thread [Thread] The child thread.
+				# @parameter channel [Channel] The notification channel for the child.
+				# @parameter options [Hash] Additional child options.
+				def initialize(thread, channel, **options)
+					@thread = thread
+					@status = nil
+					
+					super(channel, **options)
+				end
+				
+				# Raise {Interrupt} in the child thread.
+				def interrupt!
+					@thread.raise(Interrupt)
+				end
+				
+				# Raise {Terminate} in the child thread.
+				def terminate!
+					@thread.raise(Terminate)
+				end
+				
+				# Invoke {Thread#kill} on the child thread.
+				def kill!
+					@status ||= Status.new(:killed)
+					
+					@thread.kill
+				end
+				
+				# Raise {Restart} in the child thread.
+				def restart!
+					@thread.raise(Restart)
+				end
+				
+				# Reap the child thread.
+				# @parameter timeout [Numeric | Nil] The maximum time to wait for the thread to exit.
+				# @returns [Status | Nil] The thread status, or `nil` if the timeout expired.
+				def reap(timeout = nil)
+					if timeout
+						return nil unless @thread.join(timeout)
+					else
+						@thread.join
+					end
+					
+					unless @status
+						@status ||= @thread.value
+					end
+					
+					return @status
+				end
+				
+				# A pseudo exit-status wrapper for thread execution.
+				class Status
+					# Initialize the status.
+					# @parameter error [Object | Nil] The error that caused the child thread to fail.
+					def initialize(error = nil)
+						@error = error
+					end
+					
+					# The error that caused the child thread to fail.
+					# @attribute [Object | Nil]
+					attr :error
+					
+					# Whether the status represents a successful outcome.
+					# @returns [Boolean]
+					def success?
+						@error.nil?
+					end
+					
+					# Convert the status to a hash, suitable for serialization.
+					#
+					# @returns [Boolean | String] If the status is an error, the error message is returned, otherwise `true`.
+					def as_json(...)
+						if @error
+							@error.inspect
+						else
+							true
+						end
+					end
+					
+					# A human readable representation of the status.
+					def to_s
+						"\#<#{self.class} #{success? ? "success" : "failure"}>"
+					end
+				end
+			end
+		end
+	end
+end
diff --git a/lib/async/container/threaded/instance.rb b/lib/async/container/threaded/instance.rb
new file mode 100644
index 0000000..81fcd2e
--- /dev/null
+++ b/lib/async/container/threaded/instance.rb
@@ -0,0 +1,104 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require_relative "../notify/pipe"
+
+module Async
+	module Container
+		module Threaded
+			# Used to propagate the exit status of a child process invoked by {Instance#exec}.
+			class Exit < Exception
+				# Initialize the exit status.
+				# @parameter status [::Process::Status] The process exit status.
+				def initialize(status)
+					@status = status
+				end
+				
+				# The process exit status.
+				# @attribute [::Process::Status]
+				attr :status
+				
+				# The process exit status if it was an error.
+				# @returns [::Process::Status | Nil]
+				def error
+					unless status.success?
+						status
+					end
+				end
+			end
+			
+			# Represents a running child thread from the point of view of the child thread.
+			class Instance < Notify::Pipe
+				# Wrap an instance around the {Channel} instance from within the threaded child.
+				# @parameter channel [Channel] The channel to use for communication.
+				# @parameter name [String] The name of the thread.
+				def self.for(channel, name: nil)
+					instance = self.new(channel.out)
+					instance.name = name
+					
+					return instance
+				end
+				
+				# Initialize the child thread instance.
+				#
+				# @parameter io [IO] The IO object to use for communication with the parent.
+				def initialize(io)
+					@thread = ::Thread.current
+					
+					super
+				end
+				
+				# Generate a hash representation of the thread.
+				#
+				# @returns [Hash] The thread as a hash, including `process_id`, `thread_id`, and `name`.
+				def as_json(...)
+					{
+						process_id: ::Process.pid,
+						thread_id: @thread.object_id,
+						name: @thread.name,
+					}
+				end
+				
+				# Generate a JSON representation of the thread.
+				#
+				# @returns [String] The thread as JSON.
+				def to_json(...)
+					as_json.to_json(...)
+				end
+				
+				# Set the name of the thread.
+				# @parameter value [String] The name to set.
+				def name= value
+					@thread.name = value
+				end
+				
+				# Get the name of the thread.
+				# @returns [String]
+				def name
+					@thread.name
+				end
+				
+				# Execute a child process using {::Process.spawn}. In order to simulate {::Process.exec}, an {Exit} instance is raised to propagage exit status.
+				# This creates the illusion that this method does not return (normally).
+				def exec(*arguments, ready: true, **options)
+					# Always set up the notification pipe to be inherited by the spawned process.
+					self.before_spawn(arguments, options)
+					
+					if ready
+						self.ready!(status: "(spawn)")
+					end
+					
+					begin
+						pid = ::Process.spawn(*arguments, **options)
+					ensure
+						_, status = ::Process.wait2(pid)
+						
+						raise Exit, status
+					end
+				end
+			end
+		end
+	end
+end
diff --git a/lib/metrics/provider/async/container/generic.rb b/lib/metrics/provider/async/container/generic.rb
index aba27dd..c32b973 100644
--- a/lib/metrics/provider/async/container/generic.rb
+++ b/lib/metrics/provider/async/container/generic.rb
@@ -9,7 +9,7 @@
 Metrics::Provider(Async::Container::Generic) do
 	ASYNC_CONTAINER_GENERIC_HEALTH_CHECK_FAILED = Metrics.metric("async.container.generic.health_check_failed", :counter, description: "The number of health checks that failed.")
 	
-	protected def health_check_failed!(child, age_clock, health_check_timeout)
+	protected def health_check_failed(child, health_check_timeout)
 		ASYNC_CONTAINER_GENERIC_HEALTH_CHECK_FAILED.emit(1)
 		
 		super
diff --git a/test/async/container/channel.rb b/test/async/container/channel.rb
index 89df2a7..a721b57 100644
--- a/test/async/container/channel.rb
+++ b/test/async/container/channel.rb
@@ -59,8 +59,10 @@
 	with "timeout" do
 		let(:channel) {subject.new(timeout: 0.001)}
 		
-		it "fails gracefully on timeout" do
-			expect(channel.receive).to be_nil
+		it "does not treat timeouts as channel closure" do
+			expect do
+				channel.receive
+			end.to raise_exception(IO::TimeoutError)
 		end
 	end
 end
diff --git a/test/async/container/child.rb b/test/async/container/child.rb
new file mode 100644
index 0000000..c341bf1
--- /dev/null
+++ b/test/async/container/child.rb
@@ -0,0 +1,64 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require "async/container/child"
+
+describe Async::Container::Child do
+	class TestChild < Async::Container::Child
+		def initialize(channel)
+			super(channel)
+			
+			@reap_results = []
+			@killed = false
+		end
+		
+		attr :reap_results
+		
+		def kill!
+			@killed = true
+		end
+		
+		def killed?
+			@killed
+		end
+		
+		def reap(timeout = nil)
+			@reap_results << timeout
+			
+			if timeout
+				nil
+			else
+				:status
+			end
+		end
+	end
+	
+	it "kills and reaps indefinitely if the channel closes but bounded reaping times out" do
+		channel = Async::Container::Channel.new
+		child = TestChild.new(channel)
+		
+		channel.close_write
+		
+		status = child.wait(0.001)
+		
+		expect(status).to be == :status
+		expect(child).to be(:killed?)
+		expect(child.reap_results.size).to be == 2
+		expect(child.reap_results.first).to be <= Async::Container::Child::REAP_TIMEOUT
+		expect(child.reap_results.last).to be_nil
+	end
+	
+	it "uses the default reap timeout when wait has no timeout" do
+		channel = Async::Container::Channel.new
+		child = TestChild.new(channel)
+		
+		channel.close_write
+		
+		status = child.wait
+		
+		expect(status).to be == :status
+		expect(child.reap_results.first).to be == Async::Container::Child::REAP_TIMEOUT
+	end
+end
diff --git a/test/async/container/forked.rb b/test/async/container/forked.rb
index d606fe5..55cf89e 100644
--- a/test/async/container/forked.rb
+++ b/test/async/container/forked.rb
@@ -1,74 +1,15 @@
 # frozen_string_literal: true
 
 # Released under the MIT License.
-# Copyright, 2018-2026, by Samuel Williams.
-# Copyright, 2020, by Olle Jonsson.
+# Copyright, 2026, by Samuel Williams.
 
-require "async/container/best"
 require "async/container/forked"
 require "async/container/a_container"
-require "sus/fixtures/async/scheduler_context"
 
 describe Async::Container::Forked do
-	include Sus::Fixtures::Async::SchedulerContext
-	
-	let(:container) {subject.new}
-	
 	it_behaves_like Async::Container::AContainer
 	
-	it "can restart child" do
-		trigger = IO.pipe
-		pids = IO.pipe
-		
-		thread = Thread.new do
-			container.async(restart: true) do
-				trigger.first.gets
-				pids.last.puts Process.pid.to_s
-			end
-			
-			container.wait
-		end
-		
-		3.times do
-			trigger.last.puts "die"
-			_child_pid = pids.first.gets
-		end
-		
-		thread.kill
-		thread.join
-		
-		expect(container.statistics.spawns).to be == 1
-		expect(container.statistics.restarts).to be == 2
-	end
-	
-	it "can handle interrupts" do
-		finished = IO.pipe
-		interrupted = IO.pipe
-		
-		container.spawn(restart: true) do |instance|
-			Thread.handle_interrupt(Interrupt => :never) do
-				instance.ready!
-				
-				finished.first.gets
-			rescue ::Interrupt
-				interrupted.last.puts "incorrectly interrupted"
-			end
-		rescue ::Interrupt
-			interrupted.last.puts "correctly interrupted"
-		end
-		
-		container.wait_until_ready
-		
-		container.group.interrupt
-		sleep(0.001)
-		finished.last.puts "finished"
-		
-		expect(interrupted.first.gets).to be == "correctly interrupted\n"
-		
-		container.stop
-	end
-	
-	it "should be multiprocess" do
+	it "is multiprocess" do
 		expect(subject).to be(:multiprocess?)
 	end
-end if Async::Container.fork?
+end if ::Process.respond_to?(:fork) && ::Process.respond_to?(:setpgid)
diff --git a/test/async/container/forked/child.rb b/test/async/container/forked/child.rb
new file mode 100644
index 0000000..a647f59
--- /dev/null
+++ b/test/async/container/forked/child.rb
@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require "async/container/forked/child"
+require "async/container/a_child"
+
+describe Async::Container::Forked::Child do
+	it_behaves_like Async::Container::AChild
+	
+	it "returns a process status when killed" do
+		ready = ::IO.pipe
+		
+		child = subject.call(name: "test-child") do
+			ready.last.puts "ready"
+			sleep
+		end
+		
+		ready.first.gets
+		
+		status = child.stop(false)
+		
+		expect(status).to be_a(::Process::Status)
+		expect(status.termsig).to be == Signal.list["KILL"]
+	end
+	
+	it "returns a failed process status when the child raises an exception" do
+		child = subject.call(name: "test-child") do
+			raise "boom"
+		end
+		
+		status = child.wait
+		
+		expect(status).to be_a(::Process::Status)
+		expect(status.exitstatus).to be == 1
+	end
+end if ::Process.respond_to?(:fork) && ::Process.respond_to?(:setpgid)
diff --git a/test/async/container/generic.rb b/test/async/container/generic.rb
new file mode 100644
index 0000000..82cdf6a
--- /dev/null
+++ b/test/async/container/generic.rb
@@ -0,0 +1,155 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require "async/container/generic"
+require "async/container/threaded/child"
+
+describe Async::Container::Generic do
+	let(:container) {subject.new(Async::Container::Threaded::Child)}
+	
+	class RecordingPolicy < Async::Container::Policy
+		def initialize
+			@events = []
+		end
+		
+		attr :events
+		
+		def startup_failed(container, child, age:, timeout:, **options)
+			@events << [:startup_failed, age, timeout]
+			
+			super
+		end
+		
+		def health_check_failed(container, child, age:, timeout:, **options)
+			@events << [:health_check_failed, age, timeout]
+			
+			super
+		end
+	end
+	
+	it "spawns children and waits until they are ready" do
+		container.spawn do |instance|
+			instance.ready!(status: "ready")
+			sleep
+		end
+		
+		expect(container.wait_until_ready).to be == true
+		expect(container.size).to be == 1
+		expect(container.group.children.all?(&:ready?)).to be == true
+		
+		container.stop(false)
+		
+		expect(container).not.to be(:running?)
+		expect(container.statistics).to have_attributes(spawns: be == 1, failures: be == 0)
+	end
+	
+	it "records child failures" do
+		container.spawn do
+			raise "boom"
+		end
+		
+		container.wait
+		
+		expect(container).not.to be(:running?)
+		expect(container.statistics).to have_attributes(spawns: be == 1, failures: be == 1)
+		expect(container).to be(:failed?)
+	end
+	
+	it "does not block waiting for readiness after a child fails to start" do
+		container.spawn do
+			raise "boom"
+		end
+		
+		expect(container.wait_until_ready).to be == false
+		expect(container).not.to be(:running?)
+	end
+	
+	it "restarts children when requested" do
+		count = 0
+		
+		container.spawn(restart: true) do |instance|
+			count += 1
+			raise "boom" if count == 1
+			
+			instance.ready!
+			sleep
+		end
+		
+		container.wait_until_ready
+		container.stop(false)
+		
+		expect(container.statistics).to have_attributes(spawns: be == 2, restarts: be == 1, failures: be == 1)
+	end
+	
+	it "does not restart children while stopping" do
+		container.spawn(restart: true) do |instance|
+			instance.ready!
+			sleep
+		rescue Interrupt
+			# Graceful shutdown.
+		end
+		
+		expect(container.wait_until_ready).to be == true
+		
+		container.stop(true)
+		
+		expect(container.statistics).to have_attributes(spawns: be == 1, restarts: be == 0)
+	end
+	
+	it "supports keyed children" do
+		expect(container.spawn(key: :worker) do |instance|
+			instance.ready!
+			sleep
+		end).to be == true
+		
+		expect(container.wait_until_ready).to be == true
+		expect(container.spawn(key: :worker){sleep}).to be == false
+		expect(container[:worker]).not.to be_nil
+		
+		container.stop(false)
+	end
+	
+	it "fails startup when the child sends status messages but never becomes ready" do
+		policy = RecordingPolicy.new
+		container = subject.new(Async::Container::Threaded::Child, policy: policy)
+		
+		container.spawn(startup_timeout: 0.05) do |instance|
+			loop do
+				instance.status!("Still starting...")
+				sleep 0.01
+			end
+		end
+		
+		container.wait
+		
+		expect(container).not.to be(:running?)
+		expect(container).to be(:failed?)
+		expect(container.statistics.failures).to be == 1
+		expect(policy.events).to have_attributes(size: be == 1)
+		
+		event, age, timeout = policy.events.first
+		expect(event).to be == :startup_failed
+		expect(age).to be >= timeout
+		expect(timeout).to be == 0.05
+	end
+	
+	it "does not fail startup if the child becomes ready before the startup timeout" do
+		policy = RecordingPolicy.new
+		container = subject.new(Async::Container::Threaded::Child, policy: policy)
+		
+		container.spawn(startup_timeout: 1.0) do |instance|
+			instance.status!("Starting...")
+			sleep 0.01
+			instance.ready!
+		end
+		
+		container.wait
+		
+		expect(container).not.to be(:running?)
+		expect(container).not.to be(:failed?)
+		expect(container.statistics.failures).to be == 0
+		expect(policy.events).to be(:empty?)
+	end
+end
diff --git a/test/async/container/group.rb b/test/async/container/group.rb
index 11cbfe5..1837a0e 100644
--- a/test/async/container/group.rb
+++ b/test/async/container/group.rb
@@ -3,301 +3,164 @@
 # Released under the MIT License.
 # Copyright, 2025-2026, by Samuel Williams.
 
+require "async"
 require "async/container/group"
-require "async/container/channel"
 
 describe Async::Container::Group do
 	let(:group) {Async::Container::Group.new}
 	
-	with "#size" do
-		it "returns zero for empty group" do
-			expect(group.size).to be == 0
+	class FakeChild
+		def initialize
+			@events = []
 		end
 		
-		it "returns the number of running processes" do
-			channel1 = Async::Container::Channel.new
-			channel2 = Async::Container::Channel.new
-			
-			fiber1 = Fiber.new{Fiber.yield}
-			fiber2 = Fiber.new{Fiber.yield}
-			
-			fiber1.resume
-			fiber2.resume
-			
-			group.running[channel1.in] = fiber1
-			group.running[channel2.in] = fiber2
-			
-			expect(group.size).to be == 2
-		end
-	end
-	
-	with "#running?" do
-		it "returns false for empty group" do
-			expect(group).not.to be(:running?)
+		attr :events
+		
+		def interrupt!
+			@events << :interrupt
 		end
 		
-		it "returns true when processes are running" do
-			channel = Async::Container::Channel.new
-			fiber = Fiber.new{Fiber.yield}
-			fiber.resume
-			
-			group.running[channel.in] = fiber
-			
-			expect(group).to be(:running?)
+		def terminate!
+			@events << :terminate
 		end
-	end
-	
-	with "#any?" do
-		it "returns false for empty group" do
-			expect(group).not.to be(:any?)
+		
+		def kill!
+			@events << :kill
 		end
 		
-		it "returns true when processes are running" do
-			channel = Async::Container::Channel.new
-			fiber = Fiber.new{Fiber.yield}
-			fiber.resume
-			
-			group.running[channel.in] = fiber
-			
-			expect(group).to be(:any?)
+		def restart!
+			@events << :restart
 		end
 	end
 	
-	with "#empty?" do
-		it "returns true for empty group" do
-			expect(group).to be(:empty?)
-		end
+	it "tracks children and emits add/remove events" do
+		child = FakeChild.new
 		
-		it "returns false when processes are running" do
-			channel = Async::Container::Channel.new
-			fiber = Fiber.new{Fiber.yield}
-			fiber.resume
+		Sync do
+			group.add(child)
+			
+			expect(group.children).to be(:include?, child)
+			expect(group.running).to be == group.children
+			expect(group.size).to be == 1
+			expect(group).to be(:running?)
+			
+			spawn = group.wait
+			expect(spawn).to be_a(Async::Container::Group::Spawn)
+			expect(spawn.child).to be == child
+			
+			group.remove(child, :status)
 			
-			group.running[channel.in] = fiber
+			exit = group.wait
+			expect(exit).to be_a(Async::Container::Group::Exit)
+			expect(exit.child).to be == child
+			expect(exit.status).to be == :status
 			
-			expect(group).not.to be(:empty?)
+			expect(group).to be(:empty?)
 		end
 	end
 	
-	with "#inspect" do
-		it "provides human-readable representation" do
-			expect(group.inspect).to be =~ /Async::Container::Group/
-			expect(group.inspect).to be =~ /running=0/
-		end
+	it "rejects duplicate children" do
+		child = FakeChild.new
 		
-		it "shows the number of running processes" do
-			channel = Async::Container::Channel.new
-			fiber = Fiber.new{Fiber.yield}
-			fiber.resume
+		Sync do
+			group.add(child)
 			
-			group.running[channel.in] = fiber
-			
-			expect(group.inspect).to be =~ /running=1/
+			expect do
+				group.add(child)
+			end.to raise_exception(ArgumentError)
 		end
 	end
 	
-	with "#health_check!" do
-		it "resumes all fibers with :health_check! message" do
-			messages = []
-			
-			2.times do
-				channel = Async::Container::Channel.new
-				fiber = Fiber.new do
-					result = Fiber.yield
-					messages << result
-				end
-				
-				fiber.resume
-				group.running[channel.in] = fiber
-			end
-			
-			group.health_check!
-			
-			expect(messages).to be == [:health_check!, :health_check!]
-		end
+	it "rejects removing missing children" do
+		child = FakeChild.new
 		
-		it "does nothing for empty group" do
-			expect do
-				group.health_check!
-			end.not.to raise_exception
-		end
+		expect do
+			group.remove(child)
+		end.to raise_exception(ArgumentError)
 	end
 	
-	with "#interrupt" do
-		it "resumes all fibers with Interrupt" do
-			messages = []
-			
-			2.times do
-				channel = Async::Container::Channel.new
-				fiber = Fiber.new do
-					result = Fiber.yield
-					messages << result
-				end
-				
-				fiber.resume
-				group.running[channel.in] = fiber
-			end
+	it "emits update events without mutating child state" do
+		child = FakeChild.new
+		message = {ready: true}
+		
+		Sync do
+			group.add(child)
+			group.wait
 			
-			group.interrupt
+			group.update(child, message)
 			
-			expect(messages).to be == [Async::Container::Interrupt, Async::Container::Interrupt]
+			update = group.wait
+			expect(update).to be_a(Async::Container::Group::Update)
+			expect(update.child).to be == child
+			expect(update.message).to be == message
 		end
 	end
 	
-	with "#terminate" do
-		it "resumes all fibers with Terminate" do
-			messages = []
+	it "waits until the group is empty" do
+		child = FakeChild.new
+		
+		Sync do |task|
+			group.add(child)
+			group.wait
 			
-			2.times do
-				channel = Async::Container::Channel.new
-				fiber = Fiber.new do
-					result = Fiber.yield
-					messages << result
-				end
-				
-				fiber.resume
-				group.running[channel.in] = fiber
+			task.async do
+				group.remove(child)
 			end
 			
-			group.terminate
-			
-			expect(messages).to be == [Async::Container::Terminate, Async::Container::Terminate]
+			expect(group.wait_until_empty).to be == true
 		end
 	end
 	
-	with "#kill" do
-		it "resumes all fibers with Kill" do
-			messages = []
+	it "sends interrupts before killing during bounded graceful stop" do
+		child = FakeChild.new
+		
+		Sync do |task|
+			group.add(child)
+			group.wait
 			
-			2.times do
-				channel = Async::Container::Channel.new
-				fiber = Fiber.new do
-					result = Fiber.yield
-					messages << result
-				end
-				
-				fiber.resume
-				group.running[channel.in] = fiber
+			task.async do
+				sleep 0.01
+				group.remove(child)
 			end
 			
-			group.kill
+			result = group.stop(0.001)
 			
-			expect(messages).to be == [Async::Container::Kill, Async::Container::Kill]
+			expect(result).to be == true
+			expect(child.events).to be == [:interrupt, :kill]
 		end
 	end
 	
-	# Regression test for a bug where restarting a child during health check caused
-	# "RuntimeError: can't add a new key into hash during iteration"
-	# 
-	# The scenario:
-	# - A container spawns children with `restart: true` and `health_check_timeout: N`
-	# - health_check! calls @running.each_value { |fiber| fiber.resume(:health_check!) }
-	# - A resumed fiber detects health check failure and kills the child
-	# - The spawn fiber's while loop continues (restart: true) and calls wait_for with a new child
-	# - wait_for tries to add to @running while health_check! is still iterating
-	# - This used to cause: RuntimeError: can't add a new key into hash during iteration
-	it "can restart child during health_check! iteration without error" do
-		channel1 = Async::Container::Channel.new
-		channel2 = Async::Container::Channel.new
+	it "can stop gracefully when children are removed" do
+		child = FakeChild.new
 		
-		# Simulate the spawn fiber that restarts on health check failure
-		restart = true
-		fiber = Fiber.new do
-			while restart
-				result = Fiber.yield # Wait to be resumed
-				
-				if result == :health_check!
-					# Health check failed! Simulate the restart logic:
-					# The wait_for will return (simulated by breaking from this iteration)
-					# and the while loop continues, creating a new child
-					
-					# Simulate: child.kill! happens, wait_for returns
-					# Now the while loop continues and calls wait_for with new child
-					Fiber.new do
-						group.wait_for(channel2) do |msg|
-							# New child waiting
-						end
-					end.resume
-					
-					restart = false # Only do this once for the test
-				end
+		Sync do |task|
+			group.add(child)
+			group.wait
+			
+			task.async do
+				sleep 0.001
+				group.remove(child)
 			end
+			
+			expect(group.stop(true)).to be == true
+			expect(child.events).to be == [:interrupt]
 		end
-		
-		# Start the fiber and add it to @running (simulating first wait_for call)
-		fiber.resume
-		group.running[channel1.in] = fiber
-		
-		# The fix ensures this doesn't raise RuntimeError during iteration
-		expect do
-			group.health_check!
-		end.not.to raise_exception
 	end
 	
-	# Regression test with multiple children where one restarts during health check
-	it "can handle one of multiple children restarting during health check" do
-		# Create two children, both with restart capability
-		2.times do |i|
-			channel = Async::Container::Channel.new
+	it "kills immediately when graceful is false" do
+		child = FakeChild.new
+		
+		Sync do |task|
+			group.add(child)
+			group.wait
 			
-			fiber = Fiber.new do
-				iteration = 0
-				loop do
-					iteration += 1
-					result = Fiber.yield
-					
-					# First child fails health check on first iteration
-					if i == 0 && iteration == 1 && result == :health_check!
-						# Simulate health check failure and restart:
-						# Kill the old child, create new one
-						new_channel = Async::Container::Channel.new
-						
-						# This mimics what happens in spawn's while @running loop
-						# after wait_for returns due to child being killed
-						group.wait_for(new_channel) do |msg|
-							# New child process
-						end
-						
-						break # Exit this child's loop
-					end
-				end
+			task.async do
+				sleep 0.001
+				group.remove(child)
 			end
 			
-			fiber.resume
-			group.running[channel.in] = fiber
+			expect(group.stop(false)).to be == true
+			expect(child.events).to be == [:kill]
 		end
-		
-		# The fix ensures this doesn't raise RuntimeError when the first fiber restarts
-		expect do
-			group.health_check!
-		end.not.to raise_exception
-	end
-	
-	it "handles nil fiber in @running during iteration (re-entrance scenario)" do
-		# This test simulates a scenario where:
-		# 1. IO.select returns [io1, io2]
-		# 2. While resuming fiber for io1, a re-entrant call completes fiber for io2
-		# 3. When iteration continues to io2, @running[io2] is nil
-		# Without defensive check (&.), this would crash with NoMethodError
-		
-		channel1 = Async::Container::Channel.new
-		channel2 = Async::Container::Channel.new
-		
-		fiber1 = Fiber.new{group.running.delete(channel2.in)}
-		fiber2 = Fiber.new{Fiber.yield}
-		
-		fiber2.resume
-		
-		group.running[channel1.in] = fiber1
-		group.running[channel2.in] = fiber2
-		
-		# Mock select to return both channels:
-		expect(group).to receive(:select).and_return([channel1.in, channel2.in])
-		
-		# This should not crash due to &. operator:
-		group.sleep(0)
-		
-		# Verify fiber2 was removed
-		expect(group.running.key?(channel2.in)).to be == false
 	end
 end
diff --git a/test/async/container/hybrid.rb b/test/async/container/hybrid.rb
index dc62ac4..9b5edcb 100644
--- a/test/async/container/hybrid.rb
+++ b/test/async/container/hybrid.rb
@@ -1,117 +1,23 @@
 # frozen_string_literal: true
 
 # Released under the MIT License.
-# Copyright, 2019-2026, by Samuel Williams.
+# Copyright, 2026, by Samuel Williams.
 
 require "async/container/hybrid"
-require "async/container/best"
 require "async/container/a_container"
-require "sus/fixtures/async/scheduler_context"
 
 describe Async::Container::Hybrid do
-	include Sus::Fixtures::Async::SchedulerContext
-	
 	it_behaves_like Async::Container::AContainer
 	
-	it "should be multiprocess" do
-		expect(subject).to be(:multiprocess?)
-	end
-	
-	it "forcefully stops the inner threaded container on exit" do
-		stop_arguments = []
-		interrupt_count = 0
-		
-		threaded_class = Class.new
-		threaded_class.define_method(:run) do |**options, &block|
-			self
-		end
-		threaded_class.define_method(:wait_until_ready) do
-		end
-		threaded_class.define_method(:wait) do
-			@wait_count ||= 0
-			@wait_count += 1
-			
-			raise Interrupt if @wait_count == 1
-		end
-		threaded_class.define_method(:interrupt) do
-			interrupt_count += 1
-		end
-		threaded_class.define_method(:stop) do |graceful = true|
-			stop_arguments << graceful
-		end
-		
-		container_class = Class.new(subject) do
-			def spawn(**options, &block)
-				instance = Object.new
-				def instance.ready!
-				end
-				
-				block.call(instance)
-			end
-		end
-		
-		original_threaded = Async::Container.send(:remove_const, :Threaded)
-		Async::Container.const_set(:Threaded, threaded_class)
-		
-		container = container_class.new
-		container.run(count: 1, forks: 1, threads: 1) do |instance|
-			# No-op.
-		end
-		
-		expect(interrupt_count).to be == 1
-		expect(stop_arguments).to be == [false]
-	ensure
-		Async::Container.send(:remove_const, :Threaded)
-		Async::Container.const_set(:Threaded, original_threaded)
-	end
-	
-	# https://github.com/socketry/async-container/issues/58
-	#
-	# SIGINT and SIGTERM are intentionally equivalent: both are trapped in the fork and converted into `Interrupt` (see `Forked::Child.fork`), so a single signal of either kind must drain the inner threads and exit, rather than respawning them forever (the inner container has `restart: true`, the default for `async-service` managed services).
-	def exits_fork_on_single_signal(signal)
-		pids = IO.pipe
-		fork_pid = nil
-		exited = false
+	it "is a hybrid container" do
 		container = subject.new
 		
-		container.run(count: 1, forks: 1, threads: 1, restart: true) do |instance|
-			pids.last.puts(Process.pid.to_s)
-			instance.ready!
-			sleep
-		end
-		
-		container.wait_until_ready
-		
-		fork_pid = Integer(pids.first.gets)
-		
-		# Mimic a single signal delivered to the fork (e.g. memory-based worker recycling):
-		Process.kill(signal, fork_pid)
+		expect(container).to be_a(subject)
 		
-		# The fork must drain its inner threads and exit, rather than respawning them forever:
-		8.times do
-			reaped, _status = Process.waitpid2(fork_pid, Process::WNOHANG)
-			if reaped
-				exited = true
-				break
-			end
-			sleep(0.1)
-		rescue Errno::ECHILD
-			exited = true
-			break
-		end
-		
-		expect(exited).to be == true
-	ensure
-		Process.kill(:KILL, fork_pid) if fork_pid && !exited
-		container&.stop
-		pids&.each(&:close)
-	end
-	
-	it "exits the fork on a single SIGINT even when the inner container has restart: true" do
-		exits_fork_on_single_signal(:INT)
+		container.stop(false)
 	end
 	
-	it "exits the fork on a single SIGTERM even when the inner container has restart: true" do
-		exits_fork_on_single_signal(:TERM)
+	it "is multiprocess" do
+		expect(subject).to be(:multiprocess?)
 	end
-end if Async::Container.fork?
+end if ::Process.respond_to?(:fork) && ::Process.respond_to?(:setpgid)
diff --git a/test/async/container/public_interfaces.rb b/test/async/container/public_interfaces.rb
new file mode 100644
index 0000000..91d0793
--- /dev/null
+++ b/test/async/container/public_interfaces.rb
@@ -0,0 +1,22 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require "async/container"
+
+describe Async::Container do
+	it "keeps Forked, Threaded, and Hybrid as public factories" do
+		expect(Async::Container::Forked.new).to be_a(Async::Container::Generic)
+		expect(Async::Container::Threaded.new).to be_a(Async::Container::Generic)
+		expect(Async::Container::Hybrid.new).to be_a(Async::Container::Generic)
+	end
+	
+	it "can create the best container" do
+		container = Async::Container.new
+		
+		expect(container).to be_a(Async::Container::Generic)
+		
+		container.stop(false)
+	end
+end
diff --git a/test/async/container/threaded.rb b/test/async/container/threaded.rb
index ce830cd..9048431 100644
--- a/test/async/container/threaded.rb
+++ b/test/async/container/threaded.rb
@@ -1,18 +1,15 @@
 # frozen_string_literal: true
 
 # Released under the MIT License.
-# Copyright, 2018-2024, by Samuel Williams.
+# Copyright, 2026, by Samuel Williams.
 
 require "async/container/threaded"
 require "async/container/a_container"
-require "sus/fixtures/async/scheduler_context"
 
 describe Async::Container::Threaded do
-	include Sus::Fixtures::Async::SchedulerContext
-	
 	it_behaves_like Async::Container::AContainer
 	
-	it "should not be multiprocess" do
+	it "is not multiprocess" do
 		expect(subject).not.to be(:multiprocess?)
 	end
 end
diff --git a/test/async/container/threaded/child.rb b/test/async/container/threaded/child.rb
new file mode 100644
index 0000000..04ca3b2
--- /dev/null
+++ b/test/async/container/threaded/child.rb
@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require "async/container/threaded/child"
+require "async/container/a_child"
+
+describe Async::Container::Threaded::Child do
+	it_behaves_like Async::Container::AChild
+	
+	it "returns the exception as the failure status error" do
+		child = subject.call(name: "test-child") do
+			raise "boom"
+		end
+		
+		status = child.wait
+		
+		expect(status).not.to be(:success?)
+		expect(status.error).to be_a(RuntimeError)
+	end
+	
+	it "returns killed as the immediate stop status error" do
+		ready = ::Thread::Queue.new
+		
+		child = subject.call(name: "test-child") do
+			ready.push(true)
+			sleep
+		end
+		
+		ready.pop
+		
+		status = child.stop(false)
+		
+		expect(status).not.to be(:success?)
+		expect(status.error).to be == :killed
+	end
+end