diff --git a/lib/mutant/runner.rb b/lib/mutant/runner.rb
index 3116a1a2..957d2f88 100644
--- a/lib/mutant/runner.rb
+++ b/lib/mutant/runner.rb
@@ -75,7 +75,7 @@ module Mutant
         env:       config.actor_env,
         jobs:      config.jobs,
         source:    Parallel::Source::Array.new(env.mutations),
-        sink:      Sink.new(env),
+        sink:      Sink::Mutation.new(env),
         processor: env.method(:kill)
       )
     end
diff --git a/lib/mutant/runner/sink.rb b/lib/mutant/runner/sink.rb
index a4c176ab..a02aec90 100644
--- a/lib/mutant/runner/sink.rb
+++ b/lib/mutant/runner/sink.rb
@@ -1,82 +1,113 @@
 module Mutant
   class Runner
-    # Mutation result sink
+    # Abstract base class for computation sinks
     class Sink
-      include Concord.new(:env)
+      include AbstractType
 
-      # Initialize object
+      # Return sink status
       #
-      # @return [undefined]
+      # @return [Object]
       #
       # @api private
       #
-      def initialize(*)
-        super
-        @start           = Time.now
-        @subject_results = Hash.new do |_hash, subject|
-          Result::Subject.new(
-            subject:          subject,
-            mutation_results: []
-          )
-        end
-      end
+      abstract_method :status
 
-      # Return runner status
-      #
-      # @return [Status]
-      #
-      # @api private
-      #
-      def status
-        env_result
-      end
-
-      # Test if scheduling stopped
+      # Test if computation should be stopped
       #
       # @return [Boolean]
       #
       # @api private
       #
-      def stop?
-        env.config.fail_fast && !env_result.subject_results.all?(&:success?)
-      end
+      abstract_method :stop?
 
-      # Handle mutation finish
+      # Consume result
       #
-      # @param [Result::Mutation] mutation_result
+      # @param [Object] result
       #
       # @return [self]
       #
       # @api private
       #
-      def result(mutation_result)
-        mutation = mutation_result.mutation
+      abstract_method :result
 
-        original = @subject_results[mutation.subject]
+      # Mutation result sink
+      class Mutation < self
+        include Concord.new(:env)
 
-        @subject_results[mutation.subject] = original.update(
-          mutation_results: (original.mutation_results.dup << mutation_result)
-        )
+        # Initialize object
+        #
+        # @return [undefined]
+        #
+        # @api private
+        #
+        def initialize(*)
+          super
+          @start           = Time.now
+          @subject_results = Hash.new do |_hash, subject|
+            Result::Subject.new(
+              subject:          subject,
+              mutation_results: []
+            )
+          end
+        end
 
-        self
-      end
+        # Return runner status
+        #
+        # @return [Status]
+        #
+        # @api private
+        #
+        def status
+          env_result
+        end
 
-    private
+        # Test if scheduling stopped
+        #
+        # @return [Boolean]
+        #
+        # @api private
+        #
+        def stop?
+          env.config.fail_fast && !env_result.subject_results.all?(&:success?)
+        end
 
-      # Return current result
-      #
-      # @return [Result::Env]
-      #
-      # @api private
-      #
-      def env_result
-        Result::Env.new(
-          env:             env,
-          runtime:         Time.now - @start,
-          subject_results: @subject_results.values
-        )
-      end
+        # Handle mutation finish
+        #
+        # @param [Result::Mutation] mutation_result
+        #
+        # @return [self]
+        #
+        # @api private
+        #
+        def result(mutation_result)
+          mutation = mutation_result.mutation
 
-    end # Scheduler
+          original = @subject_results[mutation.subject]
+
+          @subject_results[mutation.subject] = original.update(
+            mutation_results: (original.mutation_results.dup << mutation_result)
+          )
+
+          self
+        end
+
+      private
+
+        # Return current result
+        #
+        # @return [Result::Env]
+        #
+        # @api private
+        #
+        def env_result
+          Result::Env.new(
+            env:             env,
+            runtime:         Time.now - @start,
+            subject_results: @subject_results.values
+          )
+        end
+
+      end # Mutation
+    end # Sink
   end # Runner
 end # Mutant
diff --git a/spec/unit/mutant/runner/sink_spec.rb b/spec/unit/mutant/runner/sink/mutation_spec.rb
similarity index 98%
rename from spec/unit/mutant/runner/sink_spec.rb
rename to spec/unit/mutant/runner/sink/mutation_spec.rb
index d68352ab..7607dade 100644
--- a/spec/unit/mutant/runner/sink_spec.rb
+++ b/spec/unit/mutant/runner/sink/mutation_spec.rb
@@ -1,6 +1,4 @@
-require 'spec_helper'
-
-describe Mutant::Runner::Sink do
+describe Mutant::Runner::Sink::Mutation do
   setup_shared_context
 
   shared_context 'one result' do
diff --git a/spec/unit/mutant/runner_spec.rb b/spec/unit/mutant/runner_spec.rb
index 38d08a60..9d7be896 100644
--- a/spec/unit/mutant/runner_spec.rb
+++ b/spec/unit/mutant/runner_spec.rb
@@ -37,7 +37,7 @@ RSpec.describe Mutant::Runner do
         jobs:      1,
         env:       actor_env,
         source:    Mutant::Parallel::Source::Array.new(env.mutations),
-        sink:      Mutant::Runner::Sink.new(env),
+        sink:      Mutant::Runner::Sink::Mutation.new(env),
         processor: env.method(:kill)
       )
     end