diff --git a/.gitignore b/.gitignore
index 8d2c1847..1de55709 100644
--- a/.gitignore
+++ b/.gitignore
@@ -6,6 +6,7 @@ Gemfile.lock
 .bundle
 .idea
 .ruby-version
-examples/remote_executor_db.sqlite
+vendor/
+examples/*.sqlite
 /Gemfile.local.rb
 /test.log
diff --git a/.rubocop_todo.yml b/.rubocop_todo.yml
index ed665493..34b8bb07 100644
--- a/.rubocop_todo.yml
+++ b/.rubocop_todo.yml
@@ -169,7 +169,7 @@ Metrics/CyclomaticComplexity:
 # Offense count: 135
 # Configuration parameters: CountComments, CountAsOne, ExcludedMethods.
 Metrics/MethodLength:
-  Max: 47
+  Max: 44
 
 # Offense count: 4
 # Configuration parameters: CountComments, CountAsOne.
diff --git a/examples/actors.rb b/examples/actors.rb
new file mode 100755
index 00000000..94e35907
--- /dev/null
+++ b/examples/actors.rb
@@ -0,0 +1,72 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+require_relative 'example_helper'
+
+class ExampleActor
+  def initialize
+    @value = 0
+  end
+
+  def increment
+    @value += 1
+    STDOUT.puts "Value incremented to #{@value}"
+  end
+end
+
+def server_world
+  ExampleHelper.create_world do |config|
+    config.persistence_adapter = persistence_adapter
+    config.connector           = connector
+    config.managed_actors.add('example', class: ExampleActor)
+  end
+end
+
+def client_world
+  ExampleHelper.create_world do |config|
+    config.persistence_adapter = persistence_adapter
+    config.connector           = connector
+  end
+end
+
+def db_path
+  File.expand_path('actor_remote_executor_db.sqlite', __dir__)
+end
+
+def persistence_conn_string
+  ENV['DB_CONN_STRING'] || "sqlite://#{db_path}"
+end
+
+def persistence_adapter
+  Dynflow::PersistenceAdapters::Sequel.new persistence_conn_string
+end
+
+def connector
+  proc { |world| Dynflow::Connectors::Database.new(world) }
+end
+
+command = ARGV.first || 'server'
+
+if $PROGRAM_NAME == __FILE__
+  case command
+  when 'server'
+    puts <<~MSG
+      The server is starting…. You can send the work to it by running:
+
+         #{$PROGRAM_NAME} client
+
+    MSG
+
+    world = server_world
+    world.managed_actors['example'].tell(:increment)
+    ExampleHelper.run_web_console(world)
+  when 'client'
+    world = client_world
+    100.times do |i|
+      world.message_actor('example', 'increment', [])
+    end
+  else
+    puts "Unknown command #{command}"
+    exit 1
+  end
+end
diff --git a/examples/remote_executor.rb b/examples/remote_executor.rb
index 8b3749c6..76392691 100755
--- a/examples/remote_executor.rb
+++ b/examples/remote_executor.rb
@@ -155,7 +155,7 @@ def run_client(count)
   when 'client'
     RemoteExecutorExample.run_client(ARGV[1]&.to_i)
   else
-    puts "Unknown command #{comment}"
+    puts "Unknown command #{command}"
     exit 1
   end
 elsif defined?(Sidekiq)
diff --git a/lib/dynflow/actor.rb b/lib/dynflow/actor.rb
index 84305865..9c2cf1c4 100644
--- a/lib/dynflow/actor.rb
+++ b/lib/dynflow/actor.rb
@@ -32,7 +32,7 @@ def inspect
   Concurrent::Actor::Envelope.prepend(EnvelopeBacktraceExtension)
 
   # Common parent for all the Dynflow actors defining some defaults
-  # that we preffer here.
+  # that we prefer here.
   class Actor < Concurrent::Actor::Context
     module LogWithFullBacktrace
       def log(level, message = nil, &block)
@@ -149,4 +149,58 @@ def behaviour_definition
        Concurrent::Actor::Behaviour::ErrorsOnUnknownMessage]
     end
   end
+
+  class ManagedActor < Actor
+    def initialize(implementation)
+      @implementation = implementation.new
+      # TODO: validate timer_options, start_timer?
+      if @implementation.respond_to?(:timer_options)
+        @timer = Concurrent::TimerTask.new(@implementation.timer_options) do
+          reference.tell(:tick)
+        end
+      end
+    end
+
+    def start_timer
+      @timer.execute if @timer
+    end
+
+    def stop_timer
+      @timer.shutdown if @timer
+    end
+
+    def on_message(message)
+      method, *args = message
+
+      case method
+      when :start
+        @implementation.start if @implementation.respond_to?(:start)
+      when :terminate
+        stop_timer
+        @implementation.stop if @implementation.respond_to?(:stop)
+        args.first.fulfill true
+        return
+      else
+        @implementation.send(method, *args)
+      end
+
+      return unless @implementation.respond_to?(:start_timer?)
+
+      if @implementation.start_timer?
+        start_timer
+      else
+        stop_timer
+      end
+    end
+
+    def on_envelope(envelope)
+      # TODO: "run_user_code" ?
+      # TODO: ::Foreman.settings.load_values
+      # TODO: this should be in foreman-tasks?
+      return super unless defined? ::Rails
+      ::Rails.application.executor.wrap do
+        super
+      end
+    end
+  end
 end
diff --git a/lib/dynflow/config.rb b/lib/dynflow/config.rb
index 79bd4178..4f0954ec 100644
--- a/lib/dynflow/config.rb
+++ b/lib/dynflow/config.rb
@@ -38,6 +38,10 @@ def queues
         @queues ||= @config.queues.finalized_config(self)
       end
 
+      def managed_actors
+        @config.managed_actors.actors
+      end
+
       def method_missing(name)
         return @cache[name] if @cache.key?(name)
         value = @config.send(name)
@@ -48,6 +52,19 @@ def method_missing(name)
       end
     end
 
+    class ManagedActorsConfig
+      attr_reader :actors
+
+      def initialize
+        @actors = {}
+      end
+
+      def add(name, options = {})
+        raise ArgumentError, "Actor #{name} is already defined" if @actors.key?(name)
+        @actors[name] = options
+      end
+    end
+
     class QueuesConfig
       attr_reader :queues
 
@@ -79,6 +96,10 @@ def queues
       @queues ||= QueuesConfig.new
     end
 
+    def managed_actors
+      @managed_actors ||= ManagedActorsConfig.new
+    end
+
     config_attr :logger_adapter, LoggerAdapters::Abstract do
       LoggerAdapters::Simple.new
     end
diff --git a/lib/dynflow/coordinator.rb b/lib/dynflow/coordinator.rb
index 56422771..f73fc89e 100644
--- a/lib/dynflow/coordinator.rb
+++ b/lib/dynflow/coordinator.rb
@@ -188,6 +188,17 @@ def self.inherited(klass)
       end
     end
 
+    class SingletonActorLock < LockByWorld
+      def initialize(world, actor_name)
+        super(world)
+        @data[:id] = self.class.lock_id(actor_name)
+      end
+
+      def self.lock_id(actor_name)
+        "actor:#{actor_name}"
+      end
+    end
+
     class DelayedExecutorLock < LockByWorld
       def initialize(world)
         super
diff --git a/lib/dynflow/dispatcher.rb b/lib/dynflow/dispatcher.rb
index 92273575..89e84169 100644
--- a/lib/dynflow/dispatcher.rb
+++ b/lib/dynflow/dispatcher.rb
@@ -11,6 +11,12 @@ module Dispatcher
                 optional:          Algebrick::Types::Boolean
       end
 
+      ActorMessage = type do
+        fields! actor_name: String,
+                message: String,
+                args: Array
+      end
+
       Execution = type do
         fields! execution_plan_id: String
       end
@@ -33,7 +39,7 @@ module Dispatcher
         fields! execution_plan_id: String
       end
 
-      variants Event, Execution, Ping, Status, Planning, Halt
+      variants ActorMessage, Event, Execution, Ping, Status, Planning, Halt
     end
 
     Response = Algebrick.type do
diff --git a/lib/dynflow/dispatcher/client_dispatcher.rb b/lib/dynflow/dispatcher/client_dispatcher.rb
index 8a31ec96..65f2ba1d 100644
--- a/lib/dynflow/dispatcher/client_dispatcher.rb
+++ b/lib/dynflow/dispatcher/client_dispatcher.rb
@@ -141,6 +141,9 @@ def dispatch_request(request, client_world_id, request_id)
              ignore_unknown = event.optional
              find_executor(event.execution_plan_id)
            end),
+          (on ~ActorMessage do |event|
+             find_actor_executor(event.actor_name)
+           end),
           (on ~Halt do |event|
              executor = find_executor(event.execution_plan_id)
              executor == Dispatcher::UnknownWorld ? AnyExecutor : executor
@@ -207,6 +210,15 @@ def find_executor(execution_plan_id)
         Dispatcher::UnknownWorld
       end
 
+      def find_actor_executor(actor_name)
+        # TODO: viable to check world config for singleton property before checking for lock?
+        actor_lock = @world.coordinator.find_locks(class: Coordinator::SingletonActorLock.name,
+                                                   id: "actor:#{actor_name}").first
+        return actor_lock.world_id if actor_lock
+
+        AnyExecutor
+      end
+
       def track_request(finished, request, timeout)
         id_suffix = @last_id_suffix += 1
         id = "#{@world.id}-#{id_suffix}"
@@ -240,7 +252,7 @@ def resolve_tracked_request(id, error = nil)
             (on Execution.(execution_plan_id: ~any) do |uuid|
                @world.persistence.load_execution_plan(uuid)
              end),
-            (on Event | Ping | Halt do
+            (on ActorMessage | Event | Ping | Halt do
                true
              end)
           @tracked_requests.delete(id).success! resolve_to
diff --git a/lib/dynflow/dispatcher/executor_dispatcher.rb b/lib/dynflow/dispatcher/executor_dispatcher.rb
index 7061f12f..3e0f9cd4 100644
--- a/lib/dynflow/dispatcher/executor_dispatcher.rb
+++ b/lib/dynflow/dispatcher/executor_dispatcher.rb
@@ -10,6 +10,7 @@ def initialize(world, semaphore)
 
       def handle_request(envelope)
         match(envelope.message,
+          on(ActorMessage) { handle_actor_message(envelope, envelope.message) },
           on(Planning) { perform_planning(envelope, envelope.message) },
           on(Execution) { perform_execution(envelope, envelope.message) },
           on(Event)     { perform_event(envelope, envelope.message) },
@@ -26,6 +27,21 @@ def perform_planning(envelope, planning)
         respond(envelope, Failed[e.message])
       end
 
+      def handle_actor_message(envelope, actor_message)
+        actor = @world.managed_actors[actor_message.actor_name]
+
+        unless actor
+          respond(envelope, Failed["Actor #{actor_message.actor_name} not found"])
+          return
+        end
+
+        actor.tell([actor_message.message.to_sym, *actor_message.args])
+
+        respond(envelope, Accepted)
+      rescue Dynflow::Error => e
+        respond(envelope, Failed[e.message])
+      end
+
       def perform_execution(envelope, execution)
         allocate_executor(execution.execution_plan_id, envelope.sender_id, envelope.request_id)
         execution_lock = Coordinator::ExecutionLock.new(@world, execution.execution_plan_id, envelope.sender_id, envelope.request_id)
diff --git a/lib/dynflow/rails/configuration.rb b/lib/dynflow/rails/configuration.rb
index 6a9f60ca..b6c33315 100644
--- a/lib/dynflow/rails/configuration.rb
+++ b/lib/dynflow/rails/configuration.rb
@@ -33,6 +33,8 @@ class Configuration
       # the orchestration tied to the models.
       attr_accessor :disable_active_record_actions
 
+      delegate :managed_actors, :queues, to: :world_config
+
       def initialize
         self.pool_size                = 5
         self.remote                   = ::Rails.env.production?
@@ -165,11 +167,6 @@ def world_config
         end
       end
 
-      # expose the queues definition to Rails developers
-      def queues
-        world_config.queues
-      end
-
       protected
 
       def default_sequel_adapter_options(world)
diff --git a/lib/dynflow/world.rb b/lib/dynflow/world.rb
index 2dc1856f..9530f85f 100644
--- a/lib/dynflow/world.rb
+++ b/lib/dynflow/world.rb
@@ -11,11 +11,12 @@ class World
     include Invalidation
 
     attr_reader :id, :config, :client_dispatcher, :executor_dispatcher, :executor, :connector,
-      :transaction_adapter, :logger_adapter, :coordinator,
+      :transaction_adapter, :logger_adapter, :coordinator, :managed_actors,
       :persistence, :action_classes, :subscription_index,
       :middleware, :auto_rescue, :clock, :meta, :delayed_executor, :auto_validity_check, :validity_check_timeout, :throttle_limiter,
       :termination_timeout, :terminated, :dead_letter_handler, :execution_plan_cleaner
 
+    # rubocop:disable Metrics/MethodLength
     def initialize(config)
       @config = Config::ForWorld.new(config, self)
 
@@ -51,11 +52,13 @@ def initialize(config)
       @throttle_limiter       = @config.throttle_limiter
       @terminated             = Concurrent::Promises.resolvable_event
       @termination_timeout    = @config.termination_timeout
+      @managed_actors         = {}
       calculate_subscription_index
 
       if executor
         @executor_dispatcher = spawn_and_wait(Dispatcher::ExecutorDispatcher, "executor-dispatcher", self, @config.executor_semaphore)
         executor.initialized.wait
+        spawn_managed_actors(@config.managed_actors)
       end
       update_register
       perform_validity_checks if auto_validity_check
@@ -71,6 +74,7 @@ def initialize(config)
       end
       post_initialization
     end
+    # rubocop:enable Metrics/MethodLength
 
     # performs steps once the executor is ready and invalidation of previous worls is finished.
     # Needs to be indempotent, as it can be called several times (expecially when auto_validity_check
@@ -109,6 +113,19 @@ def registered_world
       end
     end
 
+    def spawn_managed_actors(actors)
+      actors.each do |name, options|
+        begin
+          coordinator.acquire(Coordinator::SingletonActorLock.new(self, name)) if options[:singleton]
+          actor = spawn_and_wait(Dynflow::ManagedActor, "managed-actor-#{name}", options[:class])
+          actor.ask(:start).wait
+          @managed_actors[name] = actor
+        rescue Coordinator::LockError
+          logger.info "Actor #{name} already registered, skipping"
+        end
+      end
+    end
+
     def logger
       logger_adapter.dynflow_logger
     end
@@ -243,6 +260,10 @@ def event(execution_plan_id, step_id, event, done = Concurrent::Promises.resolva
       publish_request(Dispatcher::Event[execution_plan_id, step_id, event, nil, optional], done, false)
     end
 
+    def message_actor(actor_name, message, args, done = Concurrent::Promises.resolvable_future)
+      publish_request(Dispatcher::ActorMessage[actor_name, message, args], done, false)
+    end
+
     def plan_event(execution_plan_id, step_id, event, time, accepted = Concurrent::Promises.resolvable_future, optional: false)
       publish_request(Dispatcher::Event[execution_plan_id, step_id, event, time, optional], accepted, false)
     end
@@ -327,6 +348,8 @@ def start_termination
           begin
             run_before_termination_hooks
 
+            terminate_actors
+
             if delayed_executor
               logger.info "start terminating delayed_executor..."
               delayed_executor.terminate.wait(termination_timeout)
@@ -403,6 +426,15 @@ def spawn_and_wait(klass, name, *args)
       return actor
     end
 
+    def terminate_actors
+      logger.info "terminating actors..."
+      managed_actors.each do |name, actor|
+        future = Concurrent::Promises.resolvable_future
+        actor.ask([:terminate, future])
+        future.wait(termination_timeout)
+      end
+    end
+
     def terminate_executor
       return unless executor