diff --git a/examples/sub_plans_v2.rb b/examples/sub_plans_v2.rb
new file mode 100755
index 00000000..7fd90fd3
--- /dev/null
+++ b/examples/sub_plans_v2.rb
@@ -0,0 +1,75 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+example_description = <<DESC
+  Sub Plans Example
+  ===================
+
+  This example shows, how to trigger the execution plans from within a
+  run method of some action and waing for them to finish.
+
+  This is useful, when doing bulk actions, where having one big
+  execution plan would not be effective, or in case all the data are
+  not available by the time of original action planning.
+
+DESC
+
+require_relative 'example_helper'
+require_relative 'orchestrate_evented'
+
+COUNT = (ARGV[0] || 25).to_i
+
+class Foo < Dynflow::Action
+  def plan
+    plan_self
+  end
+
+  def run(event = nil)
+    case event
+    when nil
+      rng = Random.new
+      plan_event(:ping, rng.rand(25) + 1)
+      suspend
+    when :ping
+      # Finish
+    end
+  end
+end
+
+class SubPlansExample < Dynflow::Action
+  include Dynflow::Action::V2::WithSubPlans
+
+  def initiate
+    limit_concurrency_level! 3
+    super
+  end
+
+  def create_sub_plans
+    current_batch.map { |i| trigger(Foo) }
+  end
+
+  def batch_size
+    15
+  end
+
+  def batch(from, size)
+    COUNT.times.drop(from).take(size)
+  end
+
+  def total_count
+    COUNT
+  end
+end
+
+if $0 == __FILE__
+  ExampleHelper.world.action_logger.level = Logger::DEBUG
+  ExampleHelper.world
+  t1 = ExampleHelper.world.trigger(SubPlansExample)
+  puts example_description
+  puts <<-MSG.gsub(/^.*\|/, '')
+    |  Execution plans #{t1.id} with sub plans triggered
+    |  You can see the details at
+    |    #{ExampleHelper::DYNFLOW_URL}/#{t1.id}
+  MSG
+
+  ExampleHelper.run_web_console
+end
diff --git a/lib/dynflow/action.rb b/lib/dynflow/action.rb
index 8e118476..3774db6f 100644
--- a/lib/dynflow/action.rb
+++ b/lib/dynflow/action.rb
@@ -26,6 +26,7 @@ class Action < Serializable
     require 'dynflow/action/with_sub_plans'
     require 'dynflow/action/with_bulk_sub_plans'
     require 'dynflow/action/with_polling_sub_plans'
+    require 'dynflow/action/v2'
 
     def self.all_children
       children.values.inject(children.values) do |children, child|
diff --git a/lib/dynflow/action/v2.rb b/lib/dynflow/action/v2.rb
new file mode 100644
index 00000000..b2dc3ce2
--- /dev/null
+++ b/lib/dynflow/action/v2.rb
@@ -0,0 +1,9 @@
+# frozen_string_literal: true
+
+module Dynflow
+  class Action
+    module V2
+      require 'dynflow/action/v2/with_sub_plans'
+    end
+  end
+end
diff --git a/lib/dynflow/action/v2/with_sub_plans.rb b/lib/dynflow/action/v2/with_sub_plans.rb
new file mode 100644
index 00000000..1c818402
--- /dev/null
+++ b/lib/dynflow/action/v2/with_sub_plans.rb
@@ -0,0 +1,219 @@
+# frozen_string_literal: true
+
+module Dynflow::Action::V2
+  module WithSubPlans
+    include Dynflow::Action::Cancellable
+
+    DEFAULT_BATCH_SIZE = 100
+    DEFAULT_POLLING_INTERVAL = 15
+    Ping = Algebrick.atom
+
+    class SubtaskFailedException < RuntimeError
+      def backtrace
+        []
+      end
+    end
+
+    # Methods to be overridden
+    def create_sub_plans
+      raise NotImplementedError
+    end
+
+    # Should return the expected total count of tasks
+    def total_count
+      raise NotImplementedError
+    end
+
+    def batch_size
+      DEFAULT_BATCH_SIZE
+    end
+
+    # Should return a slice of size items starting from item with index from
+    def batch(from, size)
+      raise NotImplementedError
+    end
+
+    # Polling
+    def polling_interval
+      DEFAULT_POLLING_INTERVAL
+    end
+
+    # Callbacks
+    def on_finish
+    end
+
+    def on_planning_finished
+    end
+
+    def run(event = nil)
+      case event
+      when nil
+        if output[:total_count]
+          resume
+        else
+          initiate
+        end
+      when Ping
+        tick
+      when ::Dynflow::Action::Cancellable::Cancel
+        cancel!
+      when ::Dynflow::Action::Cancellable::Abort
+        abort!
+      end
+      try_to_finish || suspend_and_ping
+    end
+
+    def initiate
+      output[:planned_count] = 0
+      output[:cancelled_count] = 0
+      output[:total_count] = total_count
+      spawn_plans
+    end
+
+    def resume
+      if sub_plans.all? { |sub_plan| sub_plan.error_in_plan? }
+        output[:resumed_count] ||= 0
+        output[:resumed_count] += output[:failed_count]
+        # We're starting over and need to reset the counts
+        %w(total failed pending success).each { |key| output.delete("#{key}_count".to_sym) }
+        initiate
+      else
+        tick
+      end
+    end
+
+    def tick
+      recalculate_counts
+      spawn_plans if can_spawn_next_batch?
+    end
+
+    def suspend_and_ping
+      delay = (concurrency_limit.nil? || concurrency_limit_capacity > 0) && can_spawn_next_batch? ? nil : polling_interval
+      plan_event(Ping, delay)
+      suspend
+    end
+
+    def spawn_plans
+      sub_plans = create_sub_plans
+      sub_plans = Array[sub_plans] unless sub_plans.is_a? Array
+      increase_counts(sub_plans.count, 0)
+      on_planning_finished unless can_spawn_next_batch?
+    end
+
+    def increase_counts(planned, failed)
+      output[:planned_count] += planned + failed
+      output[:failed_count]  = output.fetch(:failed_count, 0) + failed
+      output[:pending_count] = output.fetch(:pending_count, 0) + planned
+      output[:success_count] ||= 0
+    end
+
+    def try_to_finish
+      return false unless done?
+
+      check_for_errors!
+      on_finish
+      true
+    end
+
+    def done?
+      return false if can_spawn_next_batch? || !counts_set?
+
+      total_count - output[:success_count] - output[:failed_count] - output[:cancelled_count] <= 0
+    end
+
+    def run_progress
+      return 0.1 unless counts_set? && total_count > 0
+
+      sum = output.values_at(:success_count, :cancelled_count, :failed_count).reduce(:+)
+      sum.to_f / total_count
+    end
+
+    def recalculate_counts
+      total   = total_count
+      failed  = sub_plans_count('state' => %w(paused stopped), 'result' => 'error')
+      success = sub_plans_count('state' => 'stopped', 'result' => 'success')
+      output.update(:pending_count => total - failed - success,
+                    :failed_count  => failed - output.fetch(:resumed_count, 0),
+                    :success_count => success)
+    end
+
+    def counts_set?
+      output[:total_count] && output[:success_count] && output[:failed_count] && output[:pending_count]
+    end
+
+    def check_for_errors!
+      raise SubtaskFailedException.new("A sub task failed") if output[:failed_count] > 0
+    end
+
+    # Helper for creating sub plans
+    def trigger(action_class, *args)
+      world.trigger { world.plan_with_options(action_class: action_class, args: args, caller_action: self) }
+    end
+
+    # Concurrency limitting
+    def limit_concurrency_level!(level)
+      input[:dynflow] ||= {}
+      input[:dynflow][:concurrency_limit] = level
+    end
+
+    def concurrency_limit
+      input[:dynflow] ||= {}
+      input[:dynflow][:concurrency_limit]
+    end
+
+    def concurrency_limit_capacity
+      if limit = concurrency_limit
+        return limit unless counts_set?
+        capacity = limit - (output[:planned_count] - (output[:success_count] + output[:failed_count]))
+        [0, capacity].max
+      end
+    end
+
+    # Cancellation handling
+    def cancel!(force = false)
+      # Count the not-yet-planned tasks as cancelled
+      output[:cancelled_count] = total_count - output[:planned_count]
+      # Pass the cancel event to running sub plans if they can be cancelled
+      sub_plans(:state => 'running').each { |sub_plan| sub_plan.cancel(force) if sub_plan.cancellable? }
+      suspend
+    end
+
+    def abort!
+      cancel! true
+    end
+
+    # Batching
+    # Returns the items in the current batch
+    def current_batch
+      start_position = output[:planned_count]
+      size = batch_size
+      size = concurrency_limit_capacity if concurrency_limit
+      size = start_position + size > total_count ? total_count - start_position : size
+      batch(start_position, size)
+    end
+
+    def can_spawn_next_batch?
+      remaining_count > 0
+    end
+
+    def remaining_count
+      total_count - output[:cancelled_count] - output[:planned_count]
+    end
+
+    private
+
+    # Sub-plan lookup
+    def sub_plan_filter
+      { 'caller_execution_plan_id' => execution_plan_id,
+        'caller_action_id' => self.id }
+    end
+
+    def sub_plans(filter = {})
+      world.persistence.find_execution_plans(filters: sub_plan_filter.merge(filter))
+    end
+
+    def sub_plans_count(filter = {})
+      world.persistence.find_execution_plan_counts(filters: sub_plan_filter.merge(filter))
+    end
+  end
+end
diff --git a/test/v2_sub_plans_test.rb b/test/v2_sub_plans_test.rb
new file mode 100644
index 00000000..b6c644e7
--- /dev/null
+++ b/test/v2_sub_plans_test.rb
@@ -0,0 +1,189 @@
+# frozen_string_literal: true
+require_relative 'test_helper'
+require 'mocha/minitest'
+
+module Dynflow
+  module V2SubPlansTest
+    describe 'V2 sub-plans' do
+      include PlanAssertions
+      include Dynflow::Testing::Assertions
+      include Dynflow::Testing::Factories
+      include TestHelpers
+
+      let(:world) { WorldFactory.create_world }
+
+      class ChildAction < ::Dynflow::Action
+        def run
+        end
+      end
+
+      class ParentAction < ::Dynflow::Action
+        include Dynflow::Action::V2::WithSubPlans
+
+        def plan(count, concurrency_level = nil)
+          limit_concurrency_level!(concurrency_level) if concurrency_level
+          plan_self :count => count
+        end
+
+        def create_sub_plans
+          output[:batch_count] ||= 0
+          output[:batch_count] += 1
+          current_batch.map { |i| trigger(ChildAction) }
+        end
+
+        def batch(from, size)
+          (1..total_count).to_a.slice(from, size)
+        end
+
+        def batch_size
+          5
+        end
+
+        def total_count
+          input[:count]
+        end
+      end
+
+      describe 'normal operation' do
+        it 'spawns all sub-plans in one go with high-enough batch size and polls until they are done' do
+          action = create_and_plan_action ParentAction, 3
+          action.world.expects(:trigger).times(3)
+          action = run_action action
+          _(action.output['total_count']).must_equal 3
+          _(action.output['planned_count']).must_equal 3
+          _(action.output['pending_count']).must_equal 3
+
+          ping = action.world.clock.pending_pings.first
+          _(ping.what.value.event).must_equal Dynflow::Action::V2::WithSubPlans::Ping
+          _(ping.when).must_be_within_delta(Time.now + action.polling_interval, 1)
+          persistence = mock()
+          persistence.stubs(:find_execution_plan_counts).returns(0)
+          action.world.stubs(:persistence).returns(persistence)
+
+          action.world.clock.progress
+          action.world.executor.progress
+          ping = action.world.clock.pending_pings.first
+          _(ping.what.value.event).must_equal Dynflow::Action::V2::WithSubPlans::Ping
+          _(ping.when).must_be_within_delta(Time.now + action.polling_interval * 2, 1)
+
+          persistence = mock()
+          persistence.stubs(:find_execution_plan_counts).returns(0).then.returns(3)
+          action.world.stubs(:persistence).returns(persistence)
+          action.world.clock.progress
+          action.world.executor.progress
+
+          _(action.state).must_equal :success
+          _(action.done?).must_equal true
+        end
+
+        it 'spawns sub-plans in multiple batches and polls until they are done' do
+          action = create_and_plan_action ParentAction, 7
+          action.world.expects(:trigger).times(5)
+          action = run_action action
+          _(action.output['total_count']).must_equal 7
+          _(action.output['planned_count']).must_equal 5
+          _(action.output['pending_count']).must_equal 5
+
+          _(action.world.clock.pending_pings).must_be :empty?
+          _, _, event, * = action.world.executor.events_to_process.first
+          _(event).must_equal Dynflow::Action::V2::WithSubPlans::Ping
+          persistence = mock()
+          # Simulate 3 finished
+          persistence.stubs(:find_execution_plan_counts).returns(0).then.returns(3)
+          action.world.stubs(:persistence).returns(persistence)
+
+          action.world.expects(:trigger).times(2)
+          action.world.executor.progress
+
+          ping = action.world.clock.pending_pings.first
+          _(ping.what.value.event).must_equal Dynflow::Action::V2::WithSubPlans::Ping
+          _(ping.when).must_be_within_delta(Time.now + action.polling_interval, 1)
+
+          persistence.stubs(:find_execution_plan_counts).returns(0).then.returns(7)
+          action.world.stubs(:persistence).returns(persistence)
+          action.world.clock.progress
+          action.world.executor.progress
+
+          _(action.state).must_equal :success
+          _(action.done?).must_equal true
+        end
+      end
+
+      describe 'with concurrency control' do
+        include Dynflow::Testing
+
+        it 'allows storage and retrieval' do
+          action = create_and_plan_action ParentAction, 0
+          action = run_action action
+          _(action.concurrency_limit).must_be_nil
+          _(action.concurrency_limit_capacity).must_be_nil
+
+          action = create_and_plan_action ParentAction, 0, 1
+          action = run_action action
+
+          _(action.input['dynflow']['concurrency_limit']).must_equal 1
+          _(action.concurrency_limit).must_equal 1
+          _(action.concurrency_limit_capacity).must_equal 1
+        end
+
+        it 'reduces the batch size to fit within the concurrency limit' do
+          action = create_and_plan_action ParentAction, 5, 2
+
+          # Plan first 2 sub-plans
+          action.world.expects(:trigger).times(2)
+
+          action = run_action action
+          _(action.output['total_count']).must_equal 5
+          _(action.output['planned_count']).must_equal 2
+          _(action.output['pending_count']).must_equal 2
+          _(action.concurrency_limit_capacity).must_equal 0
+          _(action.output['batch_count']).must_equal 1
+
+          ping = action.world.clock.pending_pings.first
+          _(ping.what.value.event).must_equal Dynflow::Action::V2::WithSubPlans::Ping
+          _(ping.when).must_be_within_delta(Time.now + action.polling_interval, 1)
+          persistence = mock()
+          # Simulate 1 sub-plan finished
+          persistence.stubs(:find_execution_plan_counts).returns(0).then.returns(1)
+          action.world.stubs(:persistence).returns(persistence)
+
+          # Only 1 sub-plans fits into the capacity
+          action.world.expects(:trigger).times(1)
+          action.world.clock.progress
+          action.world.executor.progress
+
+          _(action.output['planned_count']).must_equal 3
+
+          persistence = mock()
+          persistence.stubs(:find_execution_plan_counts).returns(0).then.returns(2)
+          action.world.stubs(:persistence).returns(persistence)
+          action.world.expects(:trigger).times(1)
+          action.world.clock.progress
+          action.world.executor.progress
+
+          _(action.output['planned_count']).must_equal 4
+
+          persistence = mock()
+          persistence.stubs(:find_execution_plan_counts).returns(0).then.returns(4)
+          action.world.stubs(:persistence).returns(persistence)
+          action.world.expects(:trigger).times(1)
+          action.world.clock.progress
+          action.world.executor.progress
+
+          _(action.output['planned_count']).must_equal 5
+          _(action.concurrency_limit_capacity).must_equal 1
+
+          persistence = mock()
+          persistence.stubs(:find_execution_plan_counts).returns(0).then.returns(5)
+          action.world.stubs(:persistence).returns(persistence)
+          action.world.expects(:trigger).never
+          action.world.clock.progress
+          action.world.executor.progress
+          _(action.state).must_equal :success
+          _(action.done?).must_equal true
+          _(action.concurrency_limit_capacity).must_equal 2
+        end
+      end
+    end
+  end
+end