diff --git a/.fixtures.yml b/.fixtures.yml
index 76d4ce3b..8a27c039 100644
--- a/.fixtures.yml
+++ b/.fixtures.yml
@@ -1,6 +1,3 @@
 fixtures:
   symlinks:
     reboot: "#{source_dir}"
-    boltlib: "#{source_dir}/spec/fixtures/modules/bolt/bolt-modules/boltlib"
-  repositories:
-    bolt: https://github.com/puppetlabs/bolt.git
diff --git a/.rubocop_todo.yml b/.rubocop_todo.yml
index 25e68a1e..e69de29b 100644
--- a/.rubocop_todo.yml
+++ b/.rubocop_todo.yml
@@ -1,19 +0,0 @@
-# This configuration was generated by
-# `rubocop --auto-gen-config`
-# on 2018-10-08 10:49:35 +0800 using RuboCop version 0.49.1.
-# The point is for the user to remove these configuration records
-# one by one as the offenses are removed from the code base.
-# Note that changes in the inspected code, or installation of new
-# versions of RuboCop, may require this file to be generated again.
-
-# Offense count: 6
-RSpec/AnyInstance:
-  Exclude:
-    - 'spec/functions/wait_spec.rb'
-
-# Offense count: 1
-# Configuration parameters: SkipBlocks, EnforcedStyle, SupportedStyles.
-# SupportedStyles: described_class, explicit
-RSpec/DescribedClass:
-  Exclude:
-    - 'spec/functions/wait/bolt/executor_spec.rb'
diff --git a/.sync.yml b/.sync.yml
index b15b3012..7d8668c3 100644
--- a/.sync.yml
+++ b/.sync.yml
@@ -14,10 +14,6 @@ Gemfile:
       - gem: master_manipulator
       - gem: puppet-blacksmith
         version: '~> 3.4'
-  optional:
-    ':development':
-      - gem: 'bolt'
-        condition: "Gem::Version.new(RUBY_VERSION.dup) >= Gem::Version.new('2.3.0')"
 
 appveyor.yml:
   matrix_extras:
@@ -35,14 +31,5 @@ appveyor.yml:
 .gitlab-ci.yml:
   delete: true
 
-# Due to https://tickets.puppetlabs.com/browse/PDK-1199 we need to stop the symlink checks in Travis CI
-# The symlink checks are still done in Appveyor so there's no loss in coverage
-.travis.yml:
-  includes:
-    - env: CHECK="syntax lint metadata_lint check:git_ignore check:dot_underscore check:test_file rubocop"
-    - env: CHECK=parallel_spec
-    - env: PUPPET_GEM_VERSION="~> 4.0" CHECK=parallel_spec
-      rvm: 2.1.9
-
 spec/default_facts.yml:
   unmanaged: true
diff --git a/.travis.yml b/.travis.yml
index d4e30b1c..1e6c54ea 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -20,11 +20,12 @@ env:
 matrix:
   fast_finish: true
   include:
-    # PDK Update doesn't replace this line, but instead adds a new one.
     -
-      env: CHECK="syntax lint metadata_lint check:git_ignore check:dot_underscore check:test_file rubocop"
+      env: CHECK="syntax lint metadata_lint check:symlinks check:git_ignore check:dot_underscore check:test_file rubocop"
     -
       env: CHECK=parallel_spec
+    -
+      env: PUPPET_GEM_VERSION="~> 6.0" GEM_BOLT=true CHECK=parallel_spec
     -
       env: PUPPET_GEM_VERSION="~> 4.0" CHECK=parallel_spec
       rvm: 2.1.9
diff --git a/Gemfile b/Gemfile
index 7c972781..19bf5d49 100644
--- a/Gemfile
+++ b/Gemfile
@@ -33,7 +33,9 @@ group :development do
   gem "puppet-module-posix-dev-r#{minor_version}",     require: false, platforms: [:ruby]
   gem "puppet-module-win-default-r#{minor_version}",   require: false, platforms: [:mswin, :mingw, :x64_mingw]
   gem "puppet-module-win-dev-r#{minor_version}",       require: false, platforms: [:mswin, :mingw, :x64_mingw]
-  gem "bolt",                                          require: false if Gem::Version.new(RUBY_VERSION.dup) >= Gem::Version.new('2.3.0')
+  if ENV['GEM_BOLT']
+    gem 'bolt', '~> 1.3', require: false
+  end
 end
 group :system_tests do
   gem "puppet-module-posix-system-r#{minor_version}", require: false, platforms: [:ruby]
@@ -41,6 +43,10 @@ group :system_tests do
   gem "beaker-testmode_switcher", '~> 0.4',           require: false
   gem "master_manipulator",                           require: false
   gem "puppet-blacksmith", '~> 3.4',                  require: false
+  if ENV['GEM_BOLT']
+    gem 'bolt', '~> 1.3', require: false
+    gem 'beaker-task_helper', '~> 1.5', require: false
+  end
 end
 
 puppet_version = ENV['PUPPET_GEM_VERSION']
diff --git a/README.md b/README.md
index 6f640261..37a1fb98 100644
--- a/README.md
+++ b/README.md
@@ -158,9 +158,9 @@ This can take a single reason or an array of reasons.
 
 See the [Reboot when certain conditions are met](#reboot-when-certain-conditions-are-met) section for reasons why you might reboot.
 
-### Function: `reboot::wait`
+### Plan: `reboot::wait`
 
-This function is intended to be used as part of a [plan](https://puppet.com/docs/bolt/latest/writing_plans.html) and allows Bolt to wait for a server to reboot before continuing. This function has no use in normal Puppet code (outside of plans) and will not work.
+This plan is intended to be used as part of other [plans](https://puppet.com/docs/bolt/latest/writing_plans.html) and allows Bolt to wait for a server to reboot before continuing.
 
 Here is an example of using this module to reboot servers, wait for them to come back, then check the status of a service:
 
@@ -172,12 +172,8 @@ plan myapp::patch (
   # Upgrade the application
   run_task('myapp::upgrade', $servers, { 'version' => $version })
 
-  # Reboot the servers
-  run_task('reboot', $servers)
-
-  # Wait for them to come back, this app is slow to shut down so give them
-  # 5 min to shut down
-  reboot::wait($servers, { 'disconnect_wait' => 300 })
+  # Reboot the servers. This app is slow to shut down so give them 5 minutes to reboot.
+  run_plan('reboot', $servers, reconnect_timeout => 300)
 
   # Check the status of the service
   return run_task('service', $nodes, {
@@ -189,17 +185,29 @@ plan myapp::patch (
 
 #### Parameters
 
-##### `targets`
+##### `nodes`
 
 A `TargetSpec` object containing all nodes to wait for.
 
-##### `params`
+##### `message`
+
+An optional message to log when rebooting.
+
+##### `reboot_delay`
+
+How long (in seconds) to wait before shutting down. Defaults to 0, shutdown immediately.
+
+##### `disconnect_wait`
+
+How long (in seconds) to wait before checking whether the server has rebooted. Defaults to 1.
+
+##### `reconnect_timeout`
+
+How long (in seconds) to attempt to reconnect before giving up. Defaults to 180.
 
-A `Hash` of optional timing parameters, these should be specified as an `Integer` representing seconds. Available parameters are:
+##### `retry_interval`
 
-* `disconnect_wait`
-* `reconnect_wait`
-* `retry_interval`
+How long (in seconds) to wait between retries. Defaults to 1.
 
 ## Limitations
 
diff --git a/Rakefile b/Rakefile
index d4e36dad..22c287f8 100644
--- a/Rakefile
+++ b/Rakefile
@@ -3,4 +3,6 @@ require 'puppet-syntax/tasks/puppet-syntax'
 require 'puppet_blacksmith/rake_tasks' if Bundler.rubygems.find_name('puppet-blacksmith').any?
 
 PuppetLint.configuration.send('disable_relative')
+PuppetSyntax.exclude_paths << %w[plans/*]
 
+task :beaker => :spec_prep
diff --git a/lib/puppet/functions/reboot/sleep.rb b/lib/puppet/functions/reboot/sleep.rb
new file mode 100644
index 00000000..6aa008c6
--- /dev/null
+++ b/lib/puppet/functions/reboot/sleep.rb
@@ -0,0 +1,11 @@
+# Sleeps for specified number of seconds.
+Puppet::Functions.create_function(:'reboot::sleep') do
+  # @param period Time to sleep (in seconds)
+  dispatch :sleeper do
+    required_param 'Integer', :period
+  end
+
+  def sleeper(period)
+    sleep(period)
+  end
+end
diff --git a/lib/puppet/functions/reboot/wait.rb b/lib/puppet/functions/reboot/wait.rb
deleted file mode 100644
index 14e85013..00000000
--- a/lib/puppet/functions/reboot/wait.rb
+++ /dev/null
@@ -1,72 +0,0 @@
-# Waits for nodes to reboot when executed from within a plan
-#
-# This function assumes that the nodes has just been told to reboot and
-# therefore waits for it to disconnect and reconnect again.
-#
-# This has no valid use outside plans!
-Puppet::Functions.create_function(:'reboot::wait') do
-  # @param targets A TargetSpec containing all targets to wait for
-  # @param params Extra parameters defined as a hash, valid keys are:
-  #   disconnect_wait, reconnect_wait and retry_interval. All values should be
-  #   integers and represent seconds.
-  # @example Wait for some very slow nodes to reboot.
-  #   reboot::wait($nodes, { 'disconnect_wait' => 120, 'reconnect_wait' => 600 })
-  dispatch :wait do
-    required_param 'Variant[Array,Target]', :targets
-    optional_param 'Hash', :params
-  end
-
-  def wait(targets, params = { disconnect_wait: 20, reconnect_wait: 120, retry_interval: 1 })
-    # Convert to array
-    targets = [targets].flatten
-    threads = []
-
-    targets.each do |target|
-      # We need to thread this so that we can check many nodes at once, it is
-      # possible that this could cause performance issues but only at very
-      # large scales and if you've just triggered 5000 nodes to reboot at once
-      # you have bigger problems...
-      threads << Thread.new do
-        begin
-          # If the target is connected in the beginning, wait for it to disconnect,
-          # then wait for it to come back. Some server may take many minutes to shut
-          # down.
-          wait_until(params[:disconnect_wait], params[:retry_interval]) { !connected?(target) }
-
-          # Once the target has disconnected, wait for it to come back
-          wait_until(params[:reconnect_wait], params[:retry_interval]) { connected?(target) }
-        rescue StandardError
-          raise "Timed out waiting for #{target.name} to reboot"
-        end
-      end
-    end
-
-    threads.each(&:join)
-  end
-
-  def connected?(target)
-    executor  = Puppet.lookup(:bolt_executor) { nil }
-    transport = executor.transports[target.protocol].value
-
-    case transport
-    when Bolt::Transport::Orch
-      # Check if a node is connected by hitting the /inventory endpoint
-      connection = transport.get_connection('reboot_check')
-      client     = connection.instance_variable_get('@client')
-      inventory  = client.get("inventory/#{target.name}")
-      inventory['connected']
-    else
-      # Currently only Orchestrator transport is implemented, anyone wanting
-      # SSH or WinRM functionality should implement it here and submit a PR!
-      raise "Don't know how to handle #{transport.class}"
-    end
-  end
-
-  def wait_until(timeout = 10, retry_interval = 1)
-    start = Time.now
-    until yield
-      raise 'Timeout' if (Time.now - start).to_i >= timeout
-      sleep(retry_interval)
-    end
-  end
-end
diff --git a/plans/init.pp b/plans/init.pp
new file mode 100644
index 00000000..79da68b2
--- /dev/null
+++ b/plans/init.pp
@@ -0,0 +1,101 @@
+# Reboots targets and waits for them to be available again.
+#
+# @param nodes Targets to reboot.
+# @param message Message to log with the reboot (for platforms that support it).
+# @param reboot_delay How long (in seconds) to wait before rebooting. Defaults to 1.
+# @param disconnect_wait How long (in seconds) to wait before checking whether the server has rebooted. Defaults to 10.
+# @param reconnect_timeout How long (in seconds) to attempt to reconnect before giving up. Defaults to 180.
+# @param retry_interval How long (in seconds) to wait between retries. Defaults to 1.
+plan reboot (
+  TargetSpec $nodes,
+  Optional[String] $message = undef,
+  Integer[1] $reboot_delay = 1,
+  Integer[0] $disconnect_wait = 10,
+  Integer[0] $reconnect_timeout = 180,
+  Integer[0] $retry_interval = 1,
+) {
+  $targets = get_targets($nodes)
+
+  # Get last boot time
+  $begin_boot_time_results = without_default_logging() || {
+    run_task('reboot::last_boot_time', $targets)
+  }
+
+  # Reboot; catch errors here because the connection may get cut out from underneath
+  $reboot_result = run_task('reboot', $nodes, timeout => $reboot_delay, message => $message)
+
+  # Wait long enough for all targets to trigger reboot, plus disconnect_wait to allow for shutdown time.
+  $timeouts = $reboot_result.map |$result| { $result['timeout'] }
+  $wait = max($timeouts)
+  reboot::sleep($wait+$disconnect_wait)
+
+  $start_time = Timestamp()
+  # Wait for reboot in a loop
+  ## Check if we can connect; if we can retrieve last boot time.
+  ## Mark finished for targets with a new last boot time.
+  ## If we still have targets check for timeout, sleep if not done.
+  $failed = without_default_logging() || {
+    $reconnect_timeout.reduce($targets) |$down, $_| {
+      if $down.empty() {
+        break()
+      }
+
+      $plural = if $down.size() > 1 { 's' }
+      notice("Waiting: ${$down.size()} target${plural} rebooting")
+      $current_boot_time_results = run_task('reboot::last_boot_time', $down, _catch_errors => true)
+
+      # Compare boot times
+      $failed_results = $current_boot_time_results.filter |$current_boot_time_res| {
+        # If this one errored, need to check it again
+        if !$current_boot_time_res.ok() {
+          true
+        }
+        else {
+          # If this succeeded, then we have a boot time, compare it against the begin_boot_time
+          $target_name = $current_boot_time_res.target().name()
+          $begin_boot_time_res = $begin_boot_time_results.find($target_name)
+
+          # If the boot times are the same, then we need to check it again
+          $current_boot_time_res.value() == $begin_boot_time_res.value()
+        }
+      }
+
+      # $failed_results is an array of results, turn it into a ResultSet so we can
+      # extract the targets from it
+      $failed_targets = ResultSet($failed_results).targets()
+
+      # Check for timeout if we still have failed targets
+      if !$failed_targets.empty() {
+        $elapsed_time_sec = Integer(Timestamp() - $start_time)
+        if $elapsed_time_sec >= $reconnect_timeout {
+          fail_plan(
+            "Hosts failed to come up after reboot within ${reconnect_timeout} seconds: ${failed_targets}",
+            'bolt/reboot-timeout',
+            {
+              'failed_targets' => $failed_targets,
+            }
+          )
+        }
+
+        # sleep for a small time before trying again
+        reboot::sleep($retry_interval)
+
+        # wait for all targets to be available again
+        $remaining_time = $reconnect_timeout - $elapsed_time_sec
+        wait_until_available($failed_targets, wait_time => $remaining_time, retry_interval => $retry_interval)
+      }
+
+      $failed_targets
+    }
+  }
+
+  if !$failed.empty() {
+    fail_plan(
+      "Failed to reboot ${failed}",
+      'bolt/reboot-failed',
+      {
+        'failed_targets' => $failed,
+      },
+    )
+  }
+}
diff --git a/spec/acceptance/tasks/init_spec.rb b/spec/acceptance/tasks/init_spec.rb
new file mode 100644
index 00000000..e422dd16
--- /dev/null
+++ b/spec/acceptance/tasks/init_spec.rb
@@ -0,0 +1,54 @@
+require 'spec_helper_acceptance'
+require 'beaker-task_helper/inventory'
+require 'bolt_spec/run'
+require 'time'
+
+describe 'reboot task', bolt: true do
+  include Beaker::TaskHelper::Inventory
+  include BoltSpec::Run
+
+  def module_path
+    RSpec.configuration.module_path
+  end
+
+  def config
+    { 'modulepath' => module_path }
+  end
+
+  def inventory
+    hosts_to_inventory
+  end
+
+  let(:tm) { 60 }
+
+  it 'reports the last boot time' do
+    results = run_task('reboot::last_boot_time', 'agent', config: config, inventory: inventory)
+    results.each do |res|
+      expect(res).to include('status' => 'success')
+      expect(res['result']['_output']).to be
+    end
+  end
+
+  it 'reboots a target' do
+    results = run_task('reboot', 'agent', { 'timeout' => tm }, config: config, inventory: inventory)
+    results.each do |res|
+      expect(res).to include('status' => 'success')
+      expect(res['result']['status']).to eq('queued')
+      expect(res['result']['timeout']).to eq(tm)
+    end
+
+    agents.each { |agent| retry_shutdown_abort(agent) }
+  end
+
+  it 'accepts a message' do
+    results = run_task('reboot', 'agent', { 'timeout' => tm, 'message' => 'Bolt is rebooting the computer' },
+                       config: config, inventory: inventory)
+    results.each do |res|
+      expect(res).to include('status' => 'success')
+      expect(res['result']['status']).to eq('queued')
+      expect(res['result']['timeout']).to eq(tm)
+    end
+
+    agents.each { |agent| retry_shutdown_abort(agent) }
+  end
+end
diff --git a/spec/fixtures/client-tools/orchestrator.conf b/spec/fixtures/client-tools/orchestrator.conf
deleted file mode 100644
index 45e561b7..00000000
--- a/spec/fixtures/client-tools/orchestrator.conf
+++ /dev/null
@@ -1,6 +0,0 @@
-{
-  "options" : {
-    "service-url": "https://does.not.resolve:9999",
-    "environment": "production"
-  }
-}
diff --git a/spec/functions/wait/bolt/executor_spec.rb b/spec/functions/wait/bolt/executor_spec.rb
deleted file mode 100644
index a1852cd8..00000000
--- a/spec/functions/wait/bolt/executor_spec.rb
+++ /dev/null
@@ -1,60 +0,0 @@
-require 'spec_helper'
-
-# We need to make sure that the methods we need still exist as Bolt may change
-# a lot until it hits 1.0
-describe 'Bolt::Executor', if: bolt_loaded? && tasks_available? do
-  let(:executor) { Bolt::Executor.new }
-
-  before(:each) do
-    # There's no easy way to mock the config file location within the orchestrator client
-    # so instead modify the class methods to use our fixtures
-    OrchestratorClient::Config.any_instance.stubs(:puppetlabs_root).returns(fixtures_dir) # rubocop:disable RSpec/AnyInstance
-    OrchestratorClient::Config.any_instance.stubs(:user_root).returns(fixtures_dir) # rubocop:disable RSpec/AnyInstance
-  end
-
-  it 'returns transports' do
-    expect(executor).to respond_to(:transports)
-  end
-
-  describe 'transports' do
-    let(:transports) { executor.transports }
-
-    it 'returns a hash' do
-      expect(transports).to be_a(Hash)
-    end
-
-    it 'has a pcp key' do
-      expect(transports).to have_key('pcp')
-    end
-
-    it 'is able to return us a Bolt::Transport::Orch object' do
-      expect(transports['pcp']).to respond_to(:value)
-      expect(transports['pcp'].value).to be_a(Bolt::Transport::Orch)
-    end
-
-    describe 'Bolt::Transport::Orch' do
-      let(:pcp) { transports['pcp'].value }
-
-      it 'is able to generate us a connection' do
-        expect(pcp).to respond_to(:get_connection)
-        expect(pcp.get_connection('reboot_check')).to be_a(Bolt::Transport::Orch::Connection)
-      end
-
-      describe 'Bolt::Transport::Orch::Connection' do
-        let(:connection) { pcp.get_connection('reboot_check') }
-
-        it 'has the @client instance variable' do
-          expect(connection.instance_variable_get('@client')).to be_a(OrchestratorClient)
-        end
-
-        describe 'OrchestratorClient' do
-          let(:client) { connection.instance_variable_get('@client') }
-
-          it 'allows us to make queries' do
-            expect(client).to respond_to(:get)
-          end
-        end
-      end
-    end
-  end
-end
diff --git a/spec/functions/wait_spec.rb b/spec/functions/wait_spec.rb
deleted file mode 100644
index a8d95305..00000000
--- a/spec/functions/wait_spec.rb
+++ /dev/null
@@ -1,49 +0,0 @@
-require 'spec_helper'
-
-describe 'reboot::wait', if: bolt_loaded? && tasks_available? do
-  let(:executor) { Bolt::Executor.new }
-
-  around(:each) do |example|
-    Puppet[:tasks] = true
-    Puppet.features.stubs(:bolt?).returns(true)
-
-    Puppet.override(bolt_executor: executor) do
-      example.run
-    end
-  end
-
-  before(:each) do
-    # There's no easy way to mock the config file location within the orchestrator client
-    # so instead modify the class methods to use our fixtures
-    OrchestratorClient::Config.any_instance.stubs(:puppetlabs_root).returns(fixtures_dir)
-    OrchestratorClient::Config.any_instance.stubs(:user_root).returns(fixtures_dir)
-  end
-
-  context 'when using orchestrator' do
-    let(:target) { Bolt::Target.new('example.puppet.com', 'protocol' => 'pcp') }
-
-    it 'will run with a single Target' do
-      # Mock the disconnection and reconnection of a client
-      OrchestratorClient.any_instance.expects(:get).with('inventory/example.puppet.com').returns('connected' => true)
-      OrchestratorClient.any_instance.expects(:get).with('inventory/example.puppet.com').returns('connected' => false)
-      OrchestratorClient.any_instance.expects(:get).with('inventory/example.puppet.com').returns('connected' => true)
-
-      is_expected.to run.with_params(target)
-    end
-
-    it 'will run with multiple targets' do
-      targets = (1..100).map do |num|
-        Bolt::Target.new("example#{num}.puppet.com", 'protocol' => 'pcp')
-      end
-
-      # Mock the disconnection and reconnection of all clients
-      targets.each do |targ|
-        OrchestratorClient.any_instance.expects(:get).with("inventory/#{targ.name}").returns('connected' => true)
-        OrchestratorClient.any_instance.expects(:get).with("inventory/#{targ.name}").returns('connected' => false)
-        OrchestratorClient.any_instance.expects(:get).with("inventory/#{targ.name}").returns('connected' => true)
-      end
-
-      is_expected.to run.with_params(targets)
-    end
-  end
-end
diff --git a/spec/plans/init_spec.rb b/spec/plans/init_spec.rb
new file mode 100644
index 00000000..9e61c0ef
--- /dev/null
+++ b/spec/plans/init_spec.rb
@@ -0,0 +1,74 @@
+require 'spec_helper'
+
+# Tests generally use 0 timeouts to skip sleep in plans.
+describe 'reboot plan', bolt: true do
+  if ENV['GEM_BOLT']
+    include BoltSpec::Plans
+    # Fix wait_until_available, it's broken in Bolt 1.3.0.
+    BoltSpec::Plans::MockExecutor.class_eval do
+      def wait_until_available(targets, _options)
+        Bolt::ResultSet.new(targets.map { |target| Bolt::Result.new(target) })
+      end
+    end
+  end
+
+  it 'reboots a target' do
+    time_seq = [Time.now - 1, Time.now]
+    expect_task('reboot::last_boot_time').return { |targets:, **|
+      next_time = time_seq.shift
+      Bolt::ResultSet.new(targets.map { |target| Bolt::Result.new(target, message: next_time.to_s) })
+    }.be_called_times(2)
+    expect_task('reboot').always_return('status' => 'queued', 'timeout' => 0)
+
+    result = run_plan('reboot', 'nodes' => 'foo,bar', 'disconnect_wait' => 0)
+    expect(result.value).to eq(nil)
+  end
+
+  it 'reboots a target that takes awhile to reboot' do
+    reboot_time = Time.now
+    start_time = reboot_time - 1
+    time_seq = [start_time, start_time, reboot_time]
+    expect_task('reboot::last_boot_time').return { |targets:, **|
+      next_time = time_seq.shift
+      Bolt::ResultSet.new(targets.map { |target| Bolt::Result.new(target, message: next_time.to_s) })
+    }.be_called_times(3)
+    expect_task('reboot').always_return('status' => 'queued', 'timeout' => 0)
+
+    result = run_plan('reboot', 'nodes' => 'foo,bar', 'disconnect_wait' => 0)
+    expect(result.value).to eq(nil)
+  end
+
+  it 'waits until all targets have rebooted' do
+    reboot_time = Time.now
+    start_time = reboot_time - 1
+    time_seq = [[start_time, start_time], [start_time, reboot_time], [reboot_time]]
+    expect_task('reboot::last_boot_time').return { |targets:, **|
+      Bolt::ResultSet.new(targets.zip(time_seq.shift).map { |targ, time| Bolt::Result.new(targ, message: time.to_s) })
+    }.be_called_times(3)
+    expect_task('reboot').always_return('status' => 'queued', 'timeout' => 0)
+
+    result = run_plan('reboot', 'nodes' => 'foo,bar', 'disconnect_wait' => 0)
+    expect(result.value).to eq(nil)
+  end
+
+  it 'accepts extra arguments' do
+    time_seq = [Time.now - 1, Time.now]
+    expect_task('reboot::last_boot_time').return { |targets:, **|
+      next_time = time_seq.shift
+      Bolt::ResultSet.new(targets.map { |target| Bolt::Result.new(target, message: next_time.to_s) })
+    }.be_called_times(2)
+    expect_task('reboot')
+      .with_params('timeout' => 5, 'message' => 'restarting')
+      .always_return('status' => 'queued', 'timeout' => 0)
+
+    result = run_plan('reboot', 'nodes' => 'foo,bar', 'reboot_delay' => 5, 'message' => 'restarting',
+                                'disconnect_wait' => 1, 'reconnect_timeout' => 30, 'retry_interval' => 5)
+    expect(result.value).to eq(nil)
+  end
+
+  it 'errors if last_boot_time is unavailable' do
+    expect_task('reboot::last_boot_time').error_with('kind' => 'nope', 'msg' => 'could not')
+    result = run_plan('reboot', 'nodes' => 'foo,bar')
+    expect(result).not_to be_ok
+  end
+end
diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb
index e69d11d8..d550defe 100644
--- a/spec/spec_helper.rb
+++ b/spec/spec_helper.rb
@@ -1,3 +1,7 @@
+if ENV['GEM_BOLT']
+  require 'bolt_spec/plans'
+  BoltSpec::Plans.init
+end
 
 require 'puppetlabs_spec_helper/module_spec_helper'
 require 'rspec-puppet-facts'
@@ -33,6 +37,9 @@
     # by default Puppet runs at warning level
     Puppet.settings[:strict] = :warning
   end
+
+  # Skip tasks tests unless Bolt is available
+  c.filter_run_excluding(bolt: true) unless ENV['GEM_BOLT']
 end
 
 def ensure_module_defined(module_name)
diff --git a/spec/spec_helper_acceptance.rb b/spec/spec_helper_acceptance.rb
index 9ff74e07..9d7cfe04 100644
--- a/spec/spec_helper_acceptance.rb
+++ b/spec/spec_helper_acceptance.rb
@@ -25,6 +25,17 @@
   install_dev_puppet_module_on(host, options[:forge_host] ? staging : local)
 end
 
+base_dir = File.dirname(File.expand_path(__FILE__))
+
+RSpec.configure do |c|
+  # Skip tasks tests unless Bolt is available
+  c.filter_run_excluding(bolt: true) unless ENV['GEM_BOLT']
+
+  # Make modules available locally for Bolt
+  c.add_setting :module_path
+  c.module_path = File.join(base_dir, 'fixtures', 'modules')
+end
+
 require 'rubygems' # this is necessary for ruby 1.8
 require 'puppet/version'
 WINDOWS_SHUTDOWN_ABORT = 'cmd /c shutdown /a'.freeze
diff --git a/spec/spec_helper_local.rb b/spec/spec_helper_local.rb
index 7eda84ae..04a662db 100644
--- a/spec/spec_helper_local.rb
+++ b/spec/spec_helper_local.rb
@@ -3,32 +3,3 @@
 class Object
   alias must should
 end
-
-# Bolt may not available e.g. using an old version of ruby.
-# Therefore we should safely attempt to load bolt and expose
-# a method which can be used in tests to change behavior e.g.
-# skip bolt tests if bolt is not loaded.
-begin
-  # Bolt prior to 1.0 had issues with localization therefore we need
-  # Bolt 1.0 and above. Bolt 1.0 requires Puppet 6.0.0 and above. We
-  # can't express this in the Gemfile so we need to guard loading here.
-  # Only attempt to load bolt if the Puppet version constraint is met
-  if Gem::Version.new(Puppet.version) >= Gem::Version.new('6.0.0')
-    require 'bolt/executor'
-    require 'bolt/target'
-  end
-rescue LoadError # rubocop:disable Lint/HandleExceptions
-end
-
-def bolt_loaded?
-  !defined?(Bolt).nil?
-end
-
-def tasks_available?
-  # Tasks (--tasks) were introduced in Puppet Agent 5.4.0 (PUP-7898)
-  Gem::Version.new(Puppet.version) >= Gem::Version.new('5.4.0')
-end
-
-def fixtures_dir
-  @fixtures_dir_location ||= File.join(File.dirname(__FILE__), 'fixtures')
-end
diff --git a/tasks/init.json b/tasks/init.json
index 685ae1f0..0283f461 100644
--- a/tasks/init.json
+++ b/tasks/init.json
@@ -1,14 +1,20 @@
 {
   "description": "Reboots a machine",
   "supports_noop": false,
+  "input_method": "stdin",
   "parameters": {
     "timeout": {
-      "description": "Timeout before shutdown (seconds)",
+      "description": "Timeout before shutdown (seconds); enforces a minimum of 3s",
       "type": "Optional[Variant[Pattern[/^[0-9]*$/],Integer]]"
     },
     "message": {
       "description": "Shutdown message for systems that support it",
       "type": "Optional[Pattern[/^[^|&]*$/]]"
     }
-  }
+  },
+  "implementations": [
+    { "name": "init.rb", "requirements": ["puppet-agent"] },
+    { "name": "nix.sh", "requirements": ["shell"], "input_method": "environment" },
+    { "name": "win.ps1", "requirements": ["powershell"], "input_method": "powershell" }
+  ]
 }
diff --git a/tasks/init.rb b/tasks/init.rb
old mode 100644
new mode 100755
index 48a3514c..63b5f479
--- a/tasks/init.rb
+++ b/tasks/init.rb
@@ -1,3 +1,4 @@
+#!/opt/puppetlabs/puppet/bin/ruby
 require 'facter'
 require 'json'
 
@@ -5,8 +6,10 @@
 timeout  = params['timeout'].to_i || 3
 message  = params['message']
 
-def async_command(cmd)
-  wait_time = 3
+# Force a minimum timeout of 3 seconds to allow the task response to be delivered.
+timeout = 3 if timeout < 3
+
+def async_command(cmd, wait_time = nil)
   case Facter.value(:kernel)
   when 'windows'
     # This appears to be the only way to get the processes to properly detach
@@ -21,7 +24,7 @@ def async_command(cmd)
       # Detatch itself completely
       Process.daemon
       # Wait the prescribed amount of time
-      sleep wait_time
+      sleep wait_time if wait_time
       # Replace itself with the reboot command
       exec(*cmd)
     end
@@ -39,7 +42,6 @@ def shutdown_executable_windows
 end
 
 def windows_shutdown_command(params)
-  params[:timeout] = 3 if params[:timeout] < 3
   message_params = ['/c', "\"#{params[:message]}\""] if params[:message]
   [shutdown_executable_windows, '/r', '/t', params[:timeout], '/d', 'p:4:1', message_params].join(' ')
 end
@@ -50,7 +52,7 @@ def unix_shutdown_command(params)
   flags = if Facter.value(:kernel) == 'SunOS'
             ['-y', '-i', '6', '-g', params[:timeout], escaped_message]
           else
-            ['-r', "+#{params[:timeout]}", escaped_message]
+            ['-r', params[:timeout], escaped_message]
           end
   ['shutdown', flags, '</dev/null', '>/dev/null', '2>&1', '&'].flatten
 end
@@ -58,13 +60,14 @@ def unix_shutdown_command(params)
 # Actually shut down the computer
 if Facter.value(:kernel) == 'windows'
   async_command(windows_shutdown_command(timeout: timeout, message: message))
-else
-  # Round to minutes for everything but SunOS
-  unless Facter.value(:kernel) == 'SunOS'
-    minutes = (timeout / 60.0).ceil
-    timeout = minutes
-  end
+elsif Facter.value(:kernel) == 'SunOS'
   async_command(unix_shutdown_command(timeout: timeout, message: message))
+else
+  # Specify timeout in minutes, or now. Let the forked process sleep to handle seconds.
+  timeout_min = timeout / 60
+  timeout_min = (timeout_min > 0) ? "+#{timeout_min}" : 'now'
+  timeout_sec = timeout % 60
+  async_command(unix_shutdown_command(timeout: timeout_min, message: message), timeout_sec)
 end
 
 result = {
diff --git a/tasks/last_boot_time.json b/tasks/last_boot_time.json
new file mode 100644
index 00000000..9a6f7c2a
--- /dev/null
+++ b/tasks/last_boot_time.json
@@ -0,0 +1,7 @@
+{
+  "description": "Gets the last boot time of a Linux or Windows system",
+  "implementations": [
+    {"name": "last_boot_time.sh", "requirements": ["shell"]},
+    {"name": "last_boot_time.ps1", "requirements": ["powershell"]}
+  ]
+}
diff --git a/tasks/last_boot_time.ps1 b/tasks/last_boot_time.ps1
new file mode 100644
index 00000000..2f7a6543
--- /dev/null
+++ b/tasks/last_boot_time.ps1
@@ -0,0 +1,3 @@
+$boot = Get-WmiObject -Class Win32_OperatingSystem
+$dt = $boot.ConvertToDateTime($boot.LastBootUpTime)
+Write-Output "$($dt.ToShortDateString()) $($dt.ToShortTimeString())"
diff --git a/tasks/last_boot_time.sh b/tasks/last_boot_time.sh
new file mode 100755
index 00000000..8512982e
--- /dev/null
+++ b/tasks/last_boot_time.sh
@@ -0,0 +1,6 @@
+#!/bin/sh
+if [ $(uname) = Darwin ]; then
+  last -1 reboot
+else
+  last -1 -F reboot
+fi
diff --git a/tasks/nix.json b/tasks/nix.json
new file mode 100644
index 00000000..fdc18754
--- /dev/null
+++ b/tasks/nix.json
@@ -0,0 +1,16 @@
+{
+  "description": "Reboots a machine",
+  "private": true,
+  "supports_noop": false,
+  "input_method": "environment",
+  "parameters": {
+    "timeout": {
+      "description": "Timeout before shutdown (seconds)",
+      "type": "Optional[Integer[3]]"
+    },
+    "message": {
+      "description": "Shutdown message for systems that support it",
+      "type": "Optional[Pattern[/^[^|&]*$/]]"
+    }
+  }
+}
diff --git a/tasks/nix.sh b/tasks/nix.sh
new file mode 100755
index 00000000..3c197eb7
--- /dev/null
+++ b/tasks/nix.sh
@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+set -e
+
+if [ -n "$PT_timeout" ]; then
+  timeout=$PT_timeout
+else
+  timeout=3
+fi
+
+if [ -n "$PT_message" ]; then
+  message=$PT_message
+fi
+
+# Force a minimum timeout of 3 second to allow the task response to be delivered.
+if [ $timeout -lt 3 ]; then
+  timeout=3
+fi
+
+if [[ `uname -s` == 'SunOS' ]]; then
+  shutdown -y -i 6 -g $timeout $message </dev/null >/dev/null 2>&1 &
+else
+  # Linux only supports timeout in minutes. Handle the remainder with sleep.
+  timeout_min=$(($timeout/60))
+  timeout_sec=$(($timeout%60))
+  nohup bash -c "sleep $timeout_sec; shutdown -r +$timeout_min $message" </dev/null >/dev/null 2>&1 &
+fi
+
+echo "{\"status\":\"queued\",\"timeout\":$timeout}"
+
diff --git a/tasks/win.json b/tasks/win.json
new file mode 100644
index 00000000..5967dbf3
--- /dev/null
+++ b/tasks/win.json
@@ -0,0 +1,15 @@
+{
+  "description": "Reboots a machine",
+  "private": true,
+  "supports_noop": false,
+  "parameters": {
+    "timeout": {
+      "description": "Timeout before shutdown (seconds)",
+      "type": "Optional[Integer[3]]"
+    },
+    "message": {
+      "description": "Shutdown message for systems that support it",
+      "type": "Optional[Pattern[/^[^|&]*$/]]"
+    }
+  }
+}
diff --git a/tasks/win.ps1 b/tasks/win.ps1
new file mode 100644
index 00000000..55965fa5
--- /dev/null
+++ b/tasks/win.ps1
@@ -0,0 +1,33 @@
+[CmdletBinding()]
+Param(
+  [String]$message = "",
+  [Int]$timeout = 3
+)
+# If an error is encountered, the script will stop instead of the default of "Continue"
+$ErrorActionPreference = "Stop"
+
+If (Test-Path -Path $env:SYSTEMROOT\sysnative\shutdown.exe) {
+  $executable = "$env:SYSTEMROOT\sysnative\shutdown.exe"
+}
+ElseIf (Test-Path -Path $env:SYSTEMROOT\system32\shutdown.exe) {
+  $executable = "$env:SYSTEMROOT\system32\shutdown.exe"
+}
+Else {
+  $executable = "shutdown.exe"
+}
+
+# Force a minimum timeout of 3 second to allow the response to be returned.
+If ($timeout -lt 3) {
+  $timeout = 3
+}
+
+If ($message -ne "") {
+  & $executable /r /t $timeout /d p:4:1 /c $message
+}
+Else {
+  & $executable /r /t $timeout /d p:4:1
+}
+
+
+Write-Output "{""status"":""queued"",""timeout"":${timeout}}"
+