From d8afc103054ae1401876fb328c49e30af27b23e8 Mon Sep 17 00:00:00 2001
From: Jonathon Turel <jturel@gmail.com>
Date: Sat, 18 Apr 2026 22:03:35 +0000
Subject: [PATCH 1/5] A very primitive Agent implementation

---
 .gitignore                                    |  3 +-
 .rubocop_todo.yml                             |  2 +-
 examples/agent.rb                             | 76 +++++++++++++++++++
 examples/remote_executor.rb                   |  2 +-
 lib/dynflow/coordinator.rb                    | 11 +++
 lib/dynflow/dispatcher.rb                     |  8 +-
 lib/dynflow/dispatcher/client_dispatcher.rb   | 19 ++++-
 lib/dynflow/dispatcher/executor_dispatcher.rb | 22 ++++++
 lib/dynflow/world.rb                          | 29 +++++++
 9 files changed, 167 insertions(+), 5 deletions(-)
 create mode 100755 examples/agent.rb

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..c2ea0e11 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: 48
 
 # Offense count: 4
 # Configuration parameters: CountComments, CountAsOne.
diff --git a/examples/agent.rb b/examples/agent.rb
new file mode 100755
index 00000000..6a28935f
--- /dev/null
+++ b/examples/agent.rb
@@ -0,0 +1,76 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+require_relative 'example_helper'
+
+class MyEvent
+  def initialize(value)
+    @value = value
+  end
+
+  def run(current_value)
+    new_value = current_value + @value
+    # STDOUT.puts "Computed new value of #{new_value}"
+    new_value
+  end
+end
+
+def server_world
+  ExampleHelper.create_world do |config|
+    config.persistence_adapter = persistence_adapter
+    config.connector           = connector
+  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('agent_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.register_agent('example', value: 0)
+    world.agent_event('example', MyEvent, [1])
+    ExampleHelper.run_web_console(world)
+
+    puts "Final value: #{world.find_agent('example')[:instance].value}"
+  when 'client'
+    world = client_world
+    100.times do |i|
+      world.agent_event('example', MyEvent, [i])
+      puts "Sent MyEvent with value #{i} to the server"
+    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/coordinator.rb b/lib/dynflow/coordinator.rb
index 56422771..1b60fb43 100644
--- a/lib/dynflow/coordinator.rb
+++ b/lib/dynflow/coordinator.rb
@@ -188,6 +188,17 @@ def self.inherited(klass)
       end
     end
 
+    class AgentLock < LockByWorld
+      def initialize(world, agent_name)
+        super(world)
+        @data[:id] = self.class.lock_id(agent_name)
+      end
+
+      def self.lock_id(agent_name)
+        "agent:#{agent_name}"
+      end
+    end
+
     class DelayedExecutorLock < LockByWorld
       def initialize(world)
         super
diff --git a/lib/dynflow/dispatcher.rb b/lib/dynflow/dispatcher.rb
index 92273575..386eef2d 100644
--- a/lib/dynflow/dispatcher.rb
+++ b/lib/dynflow/dispatcher.rb
@@ -11,6 +11,12 @@ module Dispatcher
                 optional:          Algebrick::Types::Boolean
       end
 
+      AgentEvent = type do
+        fields! agent_name: String,
+                event: Object,
+                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 AgentEvent, 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..1fa2cdb2 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 ~AgentEvent do |event|
+             find_agent_executor(event.agent_name)
+           end),
           (on ~Halt do |event|
              executor = find_executor(event.execution_plan_id)
              executor == Dispatcher::UnknownWorld ? AnyExecutor : executor
@@ -156,6 +159,7 @@ def dispatch_request(request, client_world_id, request_id)
           log(Logger::DEBUG, message)
           return respond(envelope, Failed[message])
         end
+
         connector.send(envelope).value!
       rescue => e
         log(Logger::ERROR, e)
@@ -207,6 +211,19 @@ def find_executor(execution_plan_id)
         Dispatcher::UnknownWorld
       end
 
+      def find_agent_executor(agent_name)
+        agent_lock = @world.coordinator.find_locks(class: Coordinator::AgentLock.name,
+                                                   id: "agent:#{agent_name}").first
+        if agent_lock
+          agent_lock.world_id
+        else
+          Dispatcher::UnknownWorld
+        end
+      rescue => e
+        log(Logger::ERROR, e)
+        Dispatcher::UnknownWorld
+      end
+
       def track_request(finished, request, timeout)
         id_suffix = @last_id_suffix += 1
         id = "#{@world.id}-#{id_suffix}"
@@ -240,7 +257,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 AgentEvent | 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..9926604e 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(AgentEvent) { handle_agent_event(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,27 @@ def perform_planning(envelope, planning)
         respond(envelope, Failed[e.message])
       end
 
+      def handle_agent_event(envelope, agent_event)
+        agent = @world.find_agent(agent_event.agent_name)
+        unless agent && agent[:instance]
+          respond(envelope, Failed["Agent #{agent_event.agent_name} not found"])
+          return
+        end
+
+        instance = agent[:instance]
+        # TODO: send_off ?
+        instance.send(agent_event.event, agent_event.args) do |value, event_class, args|
+          event = event_class.new(*args)
+          event.run(value)
+        end
+        # TODO: handle agent.failed? == true
+        # TODO: conditional blocking?
+        instance.await
+        respond(envelope, Done)
+      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/world.rb b/lib/dynflow/world.rb
index 2dc1856f..35f963f6 100644
--- a/lib/dynflow/world.rb
+++ b/lib/dynflow/world.rb
@@ -70,6 +70,29 @@ def initialize(config)
         end
       end
       post_initialization
+      @agents = {}
+    end
+
+    def find_agent(name)
+      @agents[name]
+    end
+
+    def register_agent(name, value:)
+      if executor
+        begin
+          coordinator.acquire(Coordinator::AgentLock.new(self, name))
+        rescue Coordinator::LockError
+          logger.info "Agent #{name} already registered, skipping"
+          return
+        end
+        @agents[name] = {
+          default_value: value,
+          instance: Concurrent::Agent.new(value),
+        }
+      else
+        logger.info "Finding world for agent #{name}"
+        # TODO: implement this
+      end
     end
 
     # performs steps once the executor is ready and invalidation of previous worls is finished.
@@ -78,6 +101,7 @@ def initialize(config)
     def post_initialization
       @delayed_executor ||= try_spawn(:delayed_executor, Coordinator::DelayedExecutorLock)
       @execution_plan_cleaner ||= try_spawn(:execution_plan_cleaner, Coordinator::ExecutionPlanCleanerLock)
+      # TODO: is an agent executor needed?
       update_register
       @delayed_executor.start if auto_validity_check && @delayed_executor && !@delayed_executor.started?
       self.auto_execute if @config.auto_execute
@@ -243,6 +267,11 @@ 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 agent_event(agent_name, event, args, done = Concurrent::Promises.resolvable_future)
+      # Temporarily changed to wait for acceptance
+      publish_request(Dispatcher::AgentEvent[agent_name, event, args], done, true)
+    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

From d15baf46ab121caa920ec8d6e1d50a62023e1729 Mon Sep 17 00:00:00 2001
From: Jonathon Turel <jturel@gmail.com>
Date: Mon, 20 Apr 2026 23:43:23 +0000
Subject: [PATCH 2/5] Add observer; respond with Failed on agent error

---
 lib/dynflow/dispatcher/executor_dispatcher.rb | 10 ++++++----
 lib/dynflow/world.rb                          | 14 ++++++++++----
 2 files changed, 16 insertions(+), 8 deletions(-)

diff --git a/lib/dynflow/dispatcher/executor_dispatcher.rb b/lib/dynflow/dispatcher/executor_dispatcher.rb
index 9926604e..a640cd2c 100644
--- a/lib/dynflow/dispatcher/executor_dispatcher.rb
+++ b/lib/dynflow/dispatcher/executor_dispatcher.rb
@@ -35,15 +35,17 @@ def handle_agent_event(envelope, agent_event)
         end
 
         instance = agent[:instance]
+        # Calling await on a failed instance will block indefinitely until restarted
+        respond(envelope, Failed["agent was in failed state"]) && return if instance.failed?
+
         # TODO: send_off ?
-        instance.send(agent_event.event, agent_event.args) do |value, event_class, args|
+        instance.send_off(agent_event.event, agent_event.args) do |value, event_class, args|
           event = event_class.new(*args)
           event.run(value)
         end
-        # TODO: handle agent.failed? == true
         # TODO: conditional blocking?
-        instance.await
-        respond(envelope, Done)
+        # instance.await
+        respond(envelope, Accepted)
       rescue Dynflow::Error => e
         respond(envelope, Failed[e.message])
       end
diff --git a/lib/dynflow/world.rb b/lib/dynflow/world.rb
index 35f963f6..dc436fe7 100644
--- a/lib/dynflow/world.rb
+++ b/lib/dynflow/world.rb
@@ -77,7 +77,7 @@ def find_agent(name)
       @agents[name]
     end
 
-    def register_agent(name, value:)
+    def register_agent(name, value:, observers: [])
       if executor
         begin
           coordinator.acquire(Coordinator::AgentLock.new(self, name))
@@ -85,9 +85,16 @@ def register_agent(name, value:)
           logger.info "Agent #{name} already registered, skipping"
           return
         end
+
+        agent = Concurrent::Agent.new(value)
+        observers.each do |observer|
+          agent.add_observer(observer)
+        end
+
         @agents[name] = {
           default_value: value,
-          instance: Concurrent::Agent.new(value),
+          instance: agent,
+          observers: observers,
         }
       else
         logger.info "Finding world for agent #{name}"
@@ -268,8 +275,7 @@ def event(execution_plan_id, step_id, event, done = Concurrent::Promises.resolva
     end
 
     def agent_event(agent_name, event, args, done = Concurrent::Promises.resolvable_future)
-      # Temporarily changed to wait for acceptance
-      publish_request(Dispatcher::AgentEvent[agent_name, event, args], done, true)
+      publish_request(Dispatcher::AgentEvent[agent_name, event, args], done, false)
     end
 
     def plan_event(execution_plan_id, step_id, event, time, accepted = Concurrent::Promises.resolvable_future, optional: false)

From e524487cae9f18d6480acd50ab0a3ad27c7ec4f1 Mon Sep 17 00:00:00 2001
From: Jonathon Turel <jturel@gmail.com>
Date: Mon, 27 Apr 2026 22:31:51 +0000
Subject: [PATCH 3/5] agent -> actor

---
 .rubocop_todo.yml                             |  2 +-
 examples/{agent.rb => actors.rb}              | 30 ++++-----
 lib/dynflow/config.rb                         | 20 ++++++
 lib/dynflow/coordinator.rb                    | 10 +--
 lib/dynflow/dispatcher.rb                     |  8 +--
 lib/dynflow/dispatcher/client_dispatcher.rb   | 17 +++--
 lib/dynflow/dispatcher/executor_dispatcher.rb | 22 +++----
 lib/dynflow/rails/configuration.rb            |  7 +--
 lib/dynflow/world.rb                          | 63 +++++++++----------
 9 files changed, 92 insertions(+), 87 deletions(-)
 rename examples/{agent.rb => actors.rb} (68%)

diff --git a/.rubocop_todo.yml b/.rubocop_todo.yml
index c2ea0e11..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: 48
+  Max: 44
 
 # Offense count: 4
 # Configuration parameters: CountComments, CountAsOne.
diff --git a/examples/agent.rb b/examples/actors.rb
similarity index 68%
rename from examples/agent.rb
rename to examples/actors.rb
index 6a28935f..91fae0cb 100755
--- a/examples/agent.rb
+++ b/examples/actors.rb
@@ -3,15 +3,20 @@
 
 require_relative 'example_helper'
 
-class MyEvent
-  def initialize(value)
-    @value = value
+class ExampleActor < Concurrent::Actor::Context
+  def initialize
+    @value = 0
   end
 
-  def run(current_value)
-    new_value = current_value + @value
-    # STDOUT.puts "Computed new value of #{new_value}"
-    new_value
+  def on_message(message)
+    message, args = message
+    case message
+    when :increment
+      @value += 1
+      STDOUT.puts "Value incremented to #{@value}"
+    when :terminate
+      args.fulfill(true)
+    end
   end
 end
 
@@ -19,6 +24,7 @@ def server_world
   ExampleHelper.create_world do |config|
     config.persistence_adapter = persistence_adapter
     config.connector           = connector
+    config.managed_actors.add('example', ExampleActor)
   end
 end
 
@@ -30,7 +36,7 @@ def client_world
 end
 
 def db_path
-  File.expand_path('agent_remote_executor_db.sqlite', __dir__)
+  File.expand_path('actor_remote_executor_db.sqlite', __dir__)
 end
 
 def persistence_conn_string
@@ -58,16 +64,12 @@ def connector
     MSG
 
     world = server_world
-    world.register_agent('example', value: 0)
-    world.agent_event('example', MyEvent, [1])
+    world.managed_actors['example'].tell(:increment)
     ExampleHelper.run_web_console(world)
-
-    puts "Final value: #{world.find_agent('example')[:instance].value}"
   when 'client'
     world = client_world
     100.times do |i|
-      world.agent_event('example', MyEvent, [i])
-      puts "Sent MyEvent with value #{i} to the server"
+      world.message_actor('example', 'increment', [])
     end
   else
     puts "Unknown command #{command}"
diff --git a/lib/dynflow/config.rb b/lib/dynflow/config.rb
index 79bd4178..8a97564a 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,18 @@ def method_missing(name)
       end
     end
 
+    class ManagedActorsConfig
+      attr_reader :actors
+
+      def initialize
+        @actors = {}
+      end
+
+      def add(name, actor_class)
+        @actors[name] = actor_class
+      end
+    end
+
     class QueuesConfig
       attr_reader :queues
 
@@ -79,6 +95,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 1b60fb43..f10db051 100644
--- a/lib/dynflow/coordinator.rb
+++ b/lib/dynflow/coordinator.rb
@@ -188,14 +188,14 @@ def self.inherited(klass)
       end
     end
 
-    class AgentLock < LockByWorld
-      def initialize(world, agent_name)
+    class ActorLock < LockByWorld
+      def initialize(world, actor_name)
         super(world)
-        @data[:id] = self.class.lock_id(agent_name)
+        @data[:id] = self.class.lock_id(actor_name)
       end
 
-      def self.lock_id(agent_name)
-        "agent:#{agent_name}"
+      def self.lock_id(actor_name)
+        "actor:#{actor_name}"
       end
     end
 
diff --git a/lib/dynflow/dispatcher.rb b/lib/dynflow/dispatcher.rb
index 386eef2d..89e84169 100644
--- a/lib/dynflow/dispatcher.rb
+++ b/lib/dynflow/dispatcher.rb
@@ -11,9 +11,9 @@ module Dispatcher
                 optional:          Algebrick::Types::Boolean
       end
 
-      AgentEvent = type do
-        fields! agent_name: String,
-                event: Object,
+      ActorMessage = type do
+        fields! actor_name: String,
+                message: String,
                 args: Array
       end
 
@@ -39,7 +39,7 @@ module Dispatcher
         fields! execution_plan_id: String
       end
 
-      variants AgentEvent, 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 1fa2cdb2..edaa1f14 100644
--- a/lib/dynflow/dispatcher/client_dispatcher.rb
+++ b/lib/dynflow/dispatcher/client_dispatcher.rb
@@ -141,8 +141,8 @@ def dispatch_request(request, client_world_id, request_id)
              ignore_unknown = event.optional
              find_executor(event.execution_plan_id)
            end),
-          (on ~AgentEvent do |event|
-             find_agent_executor(event.agent_name)
+          (on ~ActorMessage do |event|
+             find_actor_executor(event.actor_name)
            end),
           (on ~Halt do |event|
              executor = find_executor(event.execution_plan_id)
@@ -159,7 +159,6 @@ def dispatch_request(request, client_world_id, request_id)
           log(Logger::DEBUG, message)
           return respond(envelope, Failed[message])
         end
-
         connector.send(envelope).value!
       rescue => e
         log(Logger::ERROR, e)
@@ -211,11 +210,11 @@ def find_executor(execution_plan_id)
         Dispatcher::UnknownWorld
       end
 
-      def find_agent_executor(agent_name)
-        agent_lock = @world.coordinator.find_locks(class: Coordinator::AgentLock.name,
-                                                   id: "agent:#{agent_name}").first
-        if agent_lock
-          agent_lock.world_id
+      def find_actor_executor(actor_name)
+        actor_lock = @world.coordinator.find_locks(class: Coordinator::ActorLock.name,
+                                                   id: "actor:#{actor_name}").first
+        if actor_lock
+          actor_lock.world_id
         else
           Dispatcher::UnknownWorld
         end
@@ -257,7 +256,7 @@ def resolve_tracked_request(id, error = nil)
             (on Execution.(execution_plan_id: ~any) do |uuid|
                @world.persistence.load_execution_plan(uuid)
              end),
-            (on AgentEvent | 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 a640cd2c..3e0f9cd4 100644
--- a/lib/dynflow/dispatcher/executor_dispatcher.rb
+++ b/lib/dynflow/dispatcher/executor_dispatcher.rb
@@ -10,7 +10,7 @@ def initialize(world, semaphore)
 
       def handle_request(envelope)
         match(envelope.message,
-          on(AgentEvent) { handle_agent_event(envelope, 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) },
@@ -27,24 +27,16 @@ def perform_planning(envelope, planning)
         respond(envelope, Failed[e.message])
       end
 
-      def handle_agent_event(envelope, agent_event)
-        agent = @world.find_agent(agent_event.agent_name)
-        unless agent && agent[:instance]
-          respond(envelope, Failed["Agent #{agent_event.agent_name} not found"])
+      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
 
-        instance = agent[:instance]
-        # Calling await on a failed instance will block indefinitely until restarted
-        respond(envelope, Failed["agent was in failed state"]) && return if instance.failed?
+        actor.tell([actor_message.message.to_sym, *actor_message.args])
 
-        # TODO: send_off ?
-        instance.send_off(agent_event.event, agent_event.args) do |value, event_class, args|
-          event = event_class.new(*args)
-          event.run(value)
-        end
-        # TODO: conditional blocking?
-        # instance.await
         respond(envelope, Accepted)
       rescue Dynflow::Error => e
         respond(envelope, Failed[e.message])
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 dc436fe7..d8bb1d9c 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
@@ -70,37 +73,8 @@ def initialize(config)
         end
       end
       post_initialization
-      @agents = {}
-    end
-
-    def find_agent(name)
-      @agents[name]
-    end
-
-    def register_agent(name, value:, observers: [])
-      if executor
-        begin
-          coordinator.acquire(Coordinator::AgentLock.new(self, name))
-        rescue Coordinator::LockError
-          logger.info "Agent #{name} already registered, skipping"
-          return
-        end
-
-        agent = Concurrent::Agent.new(value)
-        observers.each do |observer|
-          agent.add_observer(observer)
-        end
-
-        @agents[name] = {
-          default_value: value,
-          instance: agent,
-          observers: observers,
-        }
-      else
-        logger.info "Finding world for agent #{name}"
-        # TODO: implement this
-      end
     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
@@ -108,7 +82,6 @@ def register_agent(name, value:, observers: [])
     def post_initialization
       @delayed_executor ||= try_spawn(:delayed_executor, Coordinator::DelayedExecutorLock)
       @execution_plan_cleaner ||= try_spawn(:execution_plan_cleaner, Coordinator::ExecutionPlanCleanerLock)
-      # TODO: is an agent executor needed?
       update_register
       @delayed_executor.start if auto_validity_check && @delayed_executor && !@delayed_executor.started?
       self.auto_execute if @config.auto_execute
@@ -140,6 +113,17 @@ def registered_world
       end
     end
 
+    def spawn_managed_actors(actors)
+      actors.each do |name, klass|
+        begin
+          coordinator.acquire(Coordinator::ActorLock.new(self, name))
+          @managed_actors[name] = spawn_and_wait(klass, "managed-actor-#{name}")
+        rescue Coordinator::LockError
+          logger.info "Actor #{name} already registered, skipping"
+        end
+      end
+    end
+
     def logger
       logger_adapter.dynflow_logger
     end
@@ -274,8 +258,8 @@ 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 agent_event(agent_name, event, args, done = Concurrent::Promises.resolvable_future)
-      publish_request(Dispatcher::AgentEvent[agent_name, event, args], done, false)
+    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)
@@ -362,6 +346,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)
@@ -438,6 +424,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
 

From 355b02383743b4851ccf33f55e325f20e7bdb3ea Mon Sep 17 00:00:00 2001
From: Jonathon Turel <jturel@gmail.com>
Date: Wed, 29 Apr 2026 01:10:22 +0000
Subject: [PATCH 4/5] Singleton & non-singleton actor

---
 lib/dynflow/actor.rb                        |  2 +-
 lib/dynflow/config.rb                       |  5 +++--
 lib/dynflow/coordinator.rb                  |  2 +-
 lib/dynflow/dispatcher/client_dispatcher.rb | 15 ++++++---------
 lib/dynflow/world.rb                        |  6 +++---
 5 files changed, 14 insertions(+), 16 deletions(-)

diff --git a/lib/dynflow/actor.rb b/lib/dynflow/actor.rb
index 84305865..c4a5fdfc 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)
diff --git a/lib/dynflow/config.rb b/lib/dynflow/config.rb
index 8a97564a..4f0954ec 100644
--- a/lib/dynflow/config.rb
+++ b/lib/dynflow/config.rb
@@ -59,8 +59,9 @@ def initialize
         @actors = {}
       end
 
-      def add(name, actor_class)
-        @actors[name] = actor_class
+      def add(name, options = {})
+        raise ArgumentError, "Actor #{name} is already defined" if @actors.key?(name)
+        @actors[name] = options
       end
     end
 
diff --git a/lib/dynflow/coordinator.rb b/lib/dynflow/coordinator.rb
index f10db051..f73fc89e 100644
--- a/lib/dynflow/coordinator.rb
+++ b/lib/dynflow/coordinator.rb
@@ -188,7 +188,7 @@ def self.inherited(klass)
       end
     end
 
-    class ActorLock < LockByWorld
+    class SingletonActorLock < LockByWorld
       def initialize(world, actor_name)
         super(world)
         @data[:id] = self.class.lock_id(actor_name)
diff --git a/lib/dynflow/dispatcher/client_dispatcher.rb b/lib/dynflow/dispatcher/client_dispatcher.rb
index edaa1f14..3084927e 100644
--- a/lib/dynflow/dispatcher/client_dispatcher.rb
+++ b/lib/dynflow/dispatcher/client_dispatcher.rb
@@ -211,16 +211,13 @@ def find_executor(execution_plan_id)
       end
 
       def find_actor_executor(actor_name)
-        actor_lock = @world.coordinator.find_locks(class: Coordinator::ActorLock.name,
+        actor_lock = @world.coordinator.find_locks(class: Coordinator::SingletonActorLock.name,
                                                    id: "actor:#{actor_name}").first
-        if actor_lock
-          actor_lock.world_id
-        else
-          Dispatcher::UnknownWorld
-        end
-      rescue => e
-        log(Logger::ERROR, e)
-        Dispatcher::UnknownWorld
+        return actor_lock.world_id if actor_lock
+
+        @world.logger.info "Actor lock for #{actor_name} not found, trying to find executor among managed actors"
+
+        AnyExecutor
       end
 
       def track_request(finished, request, timeout)
diff --git a/lib/dynflow/world.rb b/lib/dynflow/world.rb
index d8bb1d9c..bd8beb83 100644
--- a/lib/dynflow/world.rb
+++ b/lib/dynflow/world.rb
@@ -114,10 +114,10 @@ def registered_world
     end
 
     def spawn_managed_actors(actors)
-      actors.each do |name, klass|
+      actors.each do |name, options|
         begin
-          coordinator.acquire(Coordinator::ActorLock.new(self, name))
-          @managed_actors[name] = spawn_and_wait(klass, "managed-actor-#{name}")
+          coordinator.acquire(Coordinator::SingletonActorLock.new(self, name)) if options[:singleton]
+          @managed_actors[name] = spawn_and_wait(options[:class], "managed-actor-#{name}")
         rescue Coordinator::LockError
           logger.info "Actor #{name} already registered, skipping"
         end

From 0efd5f3e1f2ecca6b095942ff67cc9abf572020e Mon Sep 17 00:00:00 2001
From: Jonathon Turel <jturel@gmail.com>
Date: Mon, 4 May 2026 12:49:41 +0000
Subject: [PATCH 5/5] Users don't inherit Concurrent::Actor directly

---
 examples/actors.rb                          | 16 ++----
 lib/dynflow/actor.rb                        | 54 +++++++++++++++++++++
 lib/dynflow/dispatcher/client_dispatcher.rb |  3 +-
 lib/dynflow/world.rb                        |  4 +-
 4 files changed, 63 insertions(+), 14 deletions(-)

diff --git a/examples/actors.rb b/examples/actors.rb
index 91fae0cb..94e35907 100755
--- a/examples/actors.rb
+++ b/examples/actors.rb
@@ -3,20 +3,14 @@
 
 require_relative 'example_helper'
 
-class ExampleActor < Concurrent::Actor::Context
+class ExampleActor
   def initialize
     @value = 0
   end
 
-  def on_message(message)
-    message, args = message
-    case message
-    when :increment
-      @value += 1
-      STDOUT.puts "Value incremented to #{@value}"
-    when :terminate
-      args.fulfill(true)
-    end
+  def increment
+    @value += 1
+    STDOUT.puts "Value incremented to #{@value}"
   end
 end
 
@@ -24,7 +18,7 @@ def server_world
   ExampleHelper.create_world do |config|
     config.persistence_adapter = persistence_adapter
     config.connector           = connector
-    config.managed_actors.add('example', ExampleActor)
+    config.managed_actors.add('example', class: ExampleActor)
   end
 end
 
diff --git a/lib/dynflow/actor.rb b/lib/dynflow/actor.rb
index c4a5fdfc..9c2cf1c4 100644
--- a/lib/dynflow/actor.rb
+++ b/lib/dynflow/actor.rb
@@ -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/dispatcher/client_dispatcher.rb b/lib/dynflow/dispatcher/client_dispatcher.rb
index 3084927e..65f2ba1d 100644
--- a/lib/dynflow/dispatcher/client_dispatcher.rb
+++ b/lib/dynflow/dispatcher/client_dispatcher.rb
@@ -211,12 +211,11 @@ def find_executor(execution_plan_id)
       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
 
-        @world.logger.info "Actor lock for #{actor_name} not found, trying to find executor among managed actors"
-
         AnyExecutor
       end
 
diff --git a/lib/dynflow/world.rb b/lib/dynflow/world.rb
index bd8beb83..9530f85f 100644
--- a/lib/dynflow/world.rb
+++ b/lib/dynflow/world.rb
@@ -117,7 +117,9 @@ def spawn_managed_actors(actors)
       actors.each do |name, options|
         begin
           coordinator.acquire(Coordinator::SingletonActorLock.new(self, name)) if options[:singleton]
-          @managed_actors[name] = spawn_and_wait(options[:class], "managed-actor-#{name}")
+          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