Make workflows ActiveJob-compatible

AGENTS.md

@@ -7,6 +7,7 @@ This Rails app uses a small set of preferred libraries for common integration wo
 - `r3x` is a Rails API app that acts as a Ruby-native workflow executor and automation engine.
 - The high-level split is: framework/runtime code lives in the app and `lib/r3x/`, while user-defined workflows live under `workflows/`.
 - Workflows are file-based, Git-friendly, and loaded into a database-backed runtime that uses Active Job + Solid Queue for execution and recurring scheduling.
+- Workflow classes are enqueued directly as Active Job classes so workflow code can use `ActiveJob::Continuable` and `step` on the real workflow job instance.
 - `Solid Queue` is the active job backend for app/runtime execution. Treat queueing semantics as database-backed, not Redis-backed.
 - In the current app configuration, `Solid Queue` is not wired through `config.solid_queue.connects_to`, so queue records use the same Active Record database connection as the app in the environments configured here. That means queue inserts can participate in the same database transaction as app writes.
 - If `Solid Queue` is ever moved to a separate database, or replaced with a non-database backend, revisit any code that relies on transactional integrity between app writes and job enqueueing. In that setup, `enqueue_after_transaction_commit` and related tests become important again.
@@ -15,13 +16,14 @@ This Rails app uses a small set of preferred libraries for common integration wo
 ## Codebase Map
 
 - `lib/r3x/`: core framework code for the workflow DSL, trigger types, workflow loading, registry, execution context, recurring-task config, and shared DSL helpers.
+- `lib/r3x/workflow/executor.rb`: shared workflow execution helper that resolves the trigger and builds `Workflow::Context` for a loaded workflow class.
 - `lib/r3x/dsl/`: shared DSL infrastructure, especially validation concerns and configuration errors used by workflow-declared objects.
 - `lib/r3x/trigger_manager.rb` + `lib/r3x/trigger_manager/`: trigger infrastructure — `R3x::TriggerManager::Collection` (manages workflow triggers as a hash keyed by `unique_key`) and `R3x::TriggerManager::Execution` (wraps a trigger for runtime use).
 - `app/lib/r3x/`: runtime support code such as outputs, client wrappers, and shared concerns.
 - `app/lib/r3x/client/google/credentials.rb`: shared Google credentials loader used by Gmail and Google Sheets integrations.
 - `app/lib/r3x/client/google/gmail.rb`: Gmail API client used by `R3x::Outputs::Gmail`.
 - `R3x::Client::Google` is a project namespace; when referencing the third-party Google gem namespace, use `::Google` to avoid constant collisions.
-- `app/jobs/r3x/`: job entrypoints, especially `R3x::RunWorkflowJob`, which resolves and executes workflows, and `R3x::ChangeDetectionJob`, which evaluates change-detecting triggers before enqueueing workflow runs.
+- `app/jobs/r3x/`: job entrypoints, especially `R3x::RunWorkflowJob`, which resolves a workflow key and dispatches to the workflow job class, and `R3x::ChangeDetectionJob`, which evaluates change-detecting triggers before enqueueing workflow runs.
 - `app/models/r3x/`: runtime support models such as `R3x::TriggerState` for per-trigger change-detection state.
 - `workflows/`: user workflow packs. These are not the framework itself; they are loaded by the framework.
 - `config/initializers/r3x_workflow_loader.rb`: boot-time workflow loading hook.
@@ -30,13 +32,15 @@ This Rails app uses a small set of preferred libraries for common integration wo
 ## Runtime Flow
 
 - Workflows subclass `R3x::Workflow::Base`, declare triggers via the DSL, and implement `#run(ctx)`.
+- `R3x::Workflow::Base` is also an `ApplicationJob`; its `#perform` delegates trigger/context setup to `R3x::Workflow::Executor` and then calls `#run(ctx)` on the current job instance.
 - Workflow-declared DSL objects must validate themselves before being registered; invalid DSL configuration should raise `R3x::ConfigurationError` with collected validation errors.
 - `R3x::Workflow::PackLoader` discovers workflow entrypoints named `workflow.rb` from directories listed in `R3X_WORKFLOW_PATHS`, loads them, and registers their classes in `R3x::Workflow::Registry`.
 - `R3x::RecurringTasksConfig` turns schedulable workflow triggers into Solid Queue dynamic recurring tasks via `SolidQueue::RecurringTask`. All triggers have a `unique_key` (based on type + options hash) used for identification and duplicate detection. `schedule_all!` persists dynamic tasks and sweeps stale ones.
 - Change-detecting triggers are file-defined trigger objects that provide `cron`, `unique_key`, and `detect_changes(workflow_key:, state:)`. Their durable runtime state lives in `R3x::TriggerState`.
-- `R3x::ChangeDetectionJob` loads the trigger, fetches/updates `R3x::TriggerState`, and only enqueues `R3x::RunWorkflowJob` when the trigger reports a change.
+- `R3x::ChangeDetectionJob` loads the trigger, fetches/updates `R3x::TriggerState`, and only enqueues the workflow job class itself when the trigger reports a change.
 - Because the app currently uses `Solid Queue` as a database-backed backend on the same Active Record database connection, code may intentionally rely on a database transaction covering both `TriggerState` updates and `perform_later`. Do not assume those guarantees survive a future backend or database split.
-- `R3x::RunWorkflowJob` fetches the workflow from the registry, resolves the trigger by `trigger_key`, builds a `Workflow::Context`, and calls `workflow_class.new.run(ctx)`.
+- `R3x::RunWorkflowJob` fetches the workflow from the registry and calls `workflow_class.perform_now(trigger_key, trigger_payload: ...)` for compatibility with callers that still dispatch by workflow key.
+- Known limitation: because queued workflow runs persist the concrete workflow class name, renaming or removing a workflow class across deploys can strand older queued runs with job deserialization failures. This is currently an accepted tradeoff for preserving `ActiveJob::Continuable` on the workflow job itself.
 - Trigger discovery is filesystem-backed through `lib/r3x/triggers/*.rb`, so trigger file names, constants, and supported types must stay aligned.
 
 ## Working with Workflows
@@ -71,7 +75,7 @@ bin/workflow [options] [command] [arguments]
 
 **Global options:** `-h, --help` — print usage.
 
-`ManualRunner` (`lib/r3x/workflow/manual_runner.rb`) fetches the workflow class from the registry, picks its first manual trigger (or creates a default `Triggers::Manual`), builds a `Workflow::Context`, and calls `workflow_class.new.run(ctx)`. This is the same execution path used by `R3x::RunWorkflowJob` but without Solid Queue — it runs synchronously in the current process.
+`ManualRunner` (`lib/r3x/workflow/manual_runner.rb`) fetches the workflow class from the registry and calls `workflow_class.perform_now`. Manual execution is resolved inside the workflow execution path, so workflows can still be run manually even when they only declare non-manual triggers.
 
 ### Rake equivalents
 
@@ -82,6 +86,11 @@ bin/workflow [options] [command] [arguments]
 
 The rake tasks exist for convenience in deployment scripts; prefer `bin/workflow` for interactive use because it has richer option parsing (e.g. `--dry-run`).
 
+### Operational note
+
+- When refactoring workflow class names, remember that already queued scheduled or change-detected runs may still point at the old concrete class name.
+- If a workflow class is renamed or removed, consider cleaning up pending jobs and recurring tasks created under the old class, or accept that older queued runs may fail deserialization.
+
 ## Maintenance Warning
 
 - Keep this file synchronized with the real codebase. If you change workflow loading, trigger discovery, scheduling flow, top-level directory structure, namespaces, or the framework/user-workflow boundary, update the relevant `AGENTS.md` sections in the same change.

Gemfile.lock

@@ -1,7 +1,7 @@
 GEM
   remote: https://rubygems.org/
   specs:
-    action_text-trix (2.1.17)
+    action_text-trix (2.1.18)
       railties
     actioncable (8.1.3)
       actionpack (= 8.1.3)
@@ -79,7 +79,7 @@ GEM
       public_suffix (>= 2.0.2, < 8.0)
     ast (2.4.3)
     base64 (0.3.0)
-    bigdecimal (4.0.1)
+    bigdecimal (4.1.0)
     bootsnap (1.23.0)
       msgpack (~> 1.2)
     brakeman (8.0.4)
@@ -318,7 +318,7 @@ GEM
       racc (~> 1.4)
     os (1.1.4)
     parallel (1.27.0)
-    parser (3.3.11.0)
+    parser (3.3.11.1)
       ast (~> 2.4.1)
       racc
     pg (1.6.3)
@@ -529,7 +529,7 @@ DEPENDENCIES
   webmock
 
 CHECKSUMS
-  action_text-trix (2.1.17) sha256=b44691639d77e67169dc054ceacd1edc04d44dc3e4c6a427aa155a2beb4cc951
+  action_text-trix (2.1.18) sha256=3fdb83f8bff4145d098be283cdd47ac41caf5110bfa6df4695ed7127d7fb3642
   actioncable (8.1.3) sha256=e5bc7f75e44e6a22de29c4f43176927c3a9ce4824464b74ed18d8226e75a80f0
   actionmailbox (8.1.3) sha256=df7da474eaa0e70df4ed5a6fef66eb3b3b0f2dbf7f14518deee8d77f1b4aae59
   actionmailer (8.1.3) sha256=831f724891bb70d0aaa4d76581a6321124b6a752cb655c9346aae5479318448d
@@ -544,7 +544,7 @@ CHECKSUMS
   addressable (2.8.9) sha256=cc154fcbe689711808a43601dee7b980238ce54368d23e127421753e46895485
   ast (2.4.3) sha256=954615157c1d6a382bc27d690d973195e79db7f55e9765ac7c481c60bdb4d383
   base64 (0.3.0) sha256=27337aeabad6ffae05c265c450490628ef3ebd4b67be58257393227588f5a97b
-  bigdecimal (4.0.1) sha256=8b07d3d065a9f921c80ceaea7c9d4ae596697295b584c296fe599dd0ad01c4a7
+  bigdecimal (4.1.0) sha256=6dc07767aa3dc456ccd48e7ae70a07b474e9afd7c5bc576f80bd6da5c8dd6cae
   bootsnap (1.23.0) sha256=c1254f458d58558b58be0f8eb8f6eec2821456785b7cdd1e16248e2020d3f214
   brakeman (8.0.4) sha256=7bf921fa9638544835df9aa7b3e720a9a72c0267f34f92135955edd80d4dcf6f
   builder (3.3.0) sha256=497918d2f9dca528fdca4b88d84e4ef4387256d984b8154e9d5d3fe5a9c8835f
@@ -638,7 +638,7 @@ CHECKSUMS
   nokogiri (1.19.2-x86_64-linux-musl) sha256=93128448e61a9383a30baef041bf1f5817e22f297a1d400521e90294445069a8
   os (1.1.4) sha256=57816d6a334e7bd6aed048f4b0308226c5fb027433b67d90a9ab435f35108d3f
   parallel (1.27.0) sha256=4ac151e1806b755fb4e2dc2332cbf0e54f2e24ba821ff2d3dcf86bf6dc4ae130
-  parser (3.3.11.0) sha256=fe28ccb92d9221283ce143cb3c2e4c916e0f589ee0cc2a64867d7716354f19b1
+  parser (3.3.11.1) sha256=d17ace7aabe3e72c3cc94043714be27cc6f852f104d81aa284c2281aecc65d54
   pg (1.6.3) sha256=1388d0563e13d2758c1089e35e973a3249e955c659592d10e5b77c468f628a99
   pg (1.6.3-aarch64-linux) sha256=0698ad563e02383c27510b76bf7d4cd2de19cd1d16a5013f375dd473e4be72ea
   pg (1.6.3-aarch64-linux-musl) sha256=06a75f4ea04b05140146f2a10550b8e0d9f006a79cdaf8b5b130cde40e3ecc2c

README.md

@@ -30,3 +30,10 @@ Then open http://localhost:3000/ to view the Mission Control Jobs dashboard.
 ```bash
 bin/rails test
 ```
+
+## Operational Notes
+
+- Workflow classes are enqueued directly as Active Job classes so workflows can use `ActiveJob::Continuable` and `step` on the real workflow job instance.
+- Tradeoff: queued workflow runs persist the concrete workflow class name in Solid Queue.
+- If a workflow class is renamed or removed before an older queued run executes, that older run may fail deserialization.
+- After workflow class renames/removals, clean up pending jobs or recurring tasks that still reference the old class if you need a clean queue.

app/jobs/r3x/change_detection_job.rb

@@ -19,16 +19,12 @@ def perform(workflow_key, options = nil)
 
       TriggerState.transaction do
         if result[:changed]
-          R3x::RunWorkflowJob.perform_later(
-            workflow_key,
-            trigger_key: trigger_key,
-            trigger_payload: result[:payload]
-          )
+          workflow_class.perform_later(trigger_key, trigger_payload: result[:payload])
         end
 
         trigger_state.record_check!(result)
       end
-    rescue StandardError => e
+    rescue => e
       trigger_state.record_error!(e) if defined?(trigger_state) && trigger_state&.persisted?
       raise
     end

app/jobs/r3x/run_workflow_job.rb

@@ -9,21 +9,7 @@ def perform(workflow_key, options = nil)
 
       R3x::Workflow::PackLoader.load!
       workflow_class = R3x::Workflow::Registry.fetch(workflow_key)
-      trigger = find_trigger(workflow_class: workflow_class, trigger_key: trigger_key)
-
-      execution = R3x::TriggerManager::Execution.new(
-        trigger: trigger,
-        workflow_key: workflow_key,
-        payload: trigger_payload
-      )
-
-      ctx = R3x::Workflow::Context.new(
-        trigger: execution,
-        workflow_key: workflow_key,
-        workflow_class: workflow_class
-      )
-
-      workflow_class.new.run(ctx)
+      workflow_class.perform_now(trigger_key, trigger_payload: trigger_payload)
     end
 
     private
@@ -52,15 +38,5 @@ def normalize_options_hash(options)
         raise ArgumentError, "Expected options hash, got #{options.class.name}"
       end
     end
-
-    def find_trigger(workflow_class:, trigger_key:)
-      trigger = workflow_class.triggers_by_key[trigger_key]
-
-      if trigger.nil?
-        raise ArgumentError, "Unknown trigger key '#{trigger_key}' for workflow '#{workflow_class.workflow_key}'"
-      end
-
-      trigger
-    end
   end
 end

lib/r3x/recurring_tasks_config.rb

@@ -16,7 +16,7 @@ def schedule_all!
           triggers.each do |trigger|
             key = namespaced_key(workflow_key, trigger)
             current_keys << key
-            task_options << [ key, task_options_for(workflow_key: workflow_key, trigger: trigger) ]
+            task_options << [ key, task_options_for(workflow_class: workflow_class, trigger: trigger) ]
           end
         end
 
@@ -48,7 +48,7 @@ def to_h
           triggers.each do |trigger|
             result_key = namespaced_key(workflow_key, trigger)
             result[result_key] = task_options_for(
-              workflow_key: workflow_key, trigger: trigger
+              workflow_class: workflow_class, trigger: trigger
             ).stringify_keys
           end
         end
@@ -62,13 +62,23 @@ def namespaced_key(workflow_key, trigger)
         "workflow:#{workflow_key}:#{trigger.unique_key}"
       end
 
-      def task_options_for(workflow_key:, trigger:)
-        {
-          class: trigger.change_detecting? ? "R3x::ChangeDetectionJob" : "R3x::RunWorkflowJob",
-          args: [ workflow_key, { "trigger_key" => trigger.unique_key } ],
-          schedule: trigger.cron,
-          queue: "default"
-        }
+      def task_options_for(workflow_class:, trigger:)
+        queue_name = workflow_class.new.queue_name
+        if trigger.change_detecting?
+          {
+            class: "R3x::ChangeDetectionJob",
+            args: [ workflow_class.workflow_key, { "trigger_key" => trigger.unique_key } ],
+            schedule: trigger.cron,
+            queue: queue_name
+          }
+        else
+          {
+            class: workflow_class.name,
+            args: [ trigger.unique_key ],
+            schedule: trigger.cron,
+            queue: queue_name
+          }
+        end
       end
     end
   end

lib/r3x/workflow/base.rb

@@ -1,11 +1,31 @@
 module R3x
   module Workflow
-    class Base
+    class Base < ApplicationJob
+      include ActiveJob::Continuable
       include Dsl
       include R3x::Concerns::Logger
 
+      class << self
+        def method_added(method_name)
+          if method_name == :perform && self != Base
+            raise ArgumentError, "Do not override #perform in #{name}. Override #run(ctx) instead."
+          end
+          super
+        end
+      end
+
+      def perform(trigger_key = nil, trigger_payload: nil)
+        ctx = Executor.build_context(
+          workflow_class: self.class,
+          trigger_key: trigger_key,
+          trigger_payload: trigger_payload
+        )
+
+        run(ctx)
+      end
+
       def run(ctx)
-        raise NotImplementedError, "Workflow must implement #run(ctx)"
+        raise NotImplementedError, "#{self.class.name} must implement #run(ctx)"
       end
     end
   end

lib/r3x/workflow/executor.rb

@@ -0,0 +1,46 @@
+module R3x
+  module Workflow
+    class Executor
+      def self.build_context(...)
+        new(...).build_context
+      end
+
+      def initialize(workflow_class:, trigger_key:, trigger_payload: nil)
+        @workflow_class = workflow_class
+        @trigger_key = trigger_key
+        @trigger_payload = trigger_payload
+      end
+
+      def build_context
+        Context.new(
+          trigger: TriggerManager::Execution.new(
+            trigger: resolve_trigger,
+            workflow_key: workflow_class.workflow_key,
+            payload: trigger_payload
+          ),
+          workflow_key: workflow_class.workflow_key,
+          workflow_class: workflow_class
+        )
+      end
+
+      private
+
+      attr_reader :workflow_class, :trigger_key, :trigger_payload
+
+      def resolve_trigger
+        return manual_trigger if trigger_key.nil?
+        return manual_trigger if manual_trigger.unique_key == trigger_key
+
+        workflow_class.triggers_by_key[trigger_key] || unknown_trigger!
+      end
+
+      def manual_trigger
+        workflow_class.triggers.find(&:manual?) || Triggers::Manual.new
+      end
+
+      def unknown_trigger!
+        raise ArgumentError, "Unknown trigger key '#{trigger_key}' for workflow '#{workflow_class.workflow_key}'"
+      end
+    end
+  end
+end

lib/r3x/workflow/manual_runner.rb

@@ -11,20 +11,8 @@ def initialize
 
       def run(workflow_key)
         workflow_class = Registry.fetch(workflow_key)
-        trigger = workflow_class.triggers.find(&:manual?) || Triggers::Manual.new
 
-        execution = TriggerManager::Execution.new(
-          trigger: trigger,
-          workflow_key: workflow_key
-        )
-
-        ctx = Context.new(
-          trigger: execution,
-          workflow_key: workflow_key,
-          workflow_class: workflow_class
-        )
-
-        workflow_class.new.run(ctx)
+        workflow_class.perform_now
       end
     end
   end

test/jobs/r3x/change_detection_job_test.rb

@@ -53,20 +53,18 @@ class ChangeDetectionJobTest < ActiveSupport::TestCase
         end
       )
 
-      register_change_detecting_workflow(fake_trigger)
+      workflow_class = register_change_detecting_workflow(fake_trigger)
 
-      assert_enqueued_jobs 1, only: R3x::RunWorkflowJob do
+      assert_enqueued_jobs 1, only: workflow_class do
         ChangeDetectionJob.perform_now("test_change_detecting_feed", { "trigger_key" => fake_trigger.unique_key })
       end
 
       enqueued_job = enqueued_jobs.last
-      assert_equal R3x::RunWorkflowJob, enqueued_job[:job]
-      assert_equal "test_change_detecting_feed", enqueued_job[:args][0]
-      assert_equal fake_trigger.unique_key, enqueued_job[:args][1]["trigger_key"]
-      assert_equal(
-        { "entries" => [ { "title" => "Hello", "_aj_symbol_keys" => [ "title" ] } ], "_aj_symbol_keys" => [ "entries" ] },
-        enqueued_job[:args][1]["trigger_payload"]
-      )
+      assert_equal workflow_class, enqueued_job[:job]
+      assert_equal fake_trigger.unique_key, enqueued_job[:args][0]
+      payload = enqueued_job[:args][1]["trigger_payload"]
+      assert_equal 1, payload["entries"].length
+      assert_equal "Hello", payload["entries"].first["title"]
 
       state = R3x::TriggerState.find_by!(workflow_key: "test_change_detecting_feed", trigger_key: fake_trigger.unique_key)
       assert_equal({ "cursor" => "v2" }, state.state)
@@ -81,17 +79,17 @@ class ChangeDetectionJobTest < ActiveSupport::TestCase
         end
       )
 
-      register_change_detecting_workflow(fake_trigger)
+      workflow_class = register_change_detecting_workflow(fake_trigger)
 
-      original_perform_later = R3x::RunWorkflowJob.method(:perform_later)
-      R3x::RunWorkflowJob.singleton_class.send(:define_method, :perform_later) do |*|
+      original_perform_later = workflow_class.method(:perform_later)
+      workflow_class.singleton_class.send(:define_method, :perform_later) do |*|
         raise ActiveJob::EnqueueError, "enqueue failed"
       end
 
       error = assert_raises(ActiveJob::EnqueueError) do
         ChangeDetectionJob.perform_now("test_change_detecting_feed", { "trigger_key" => fake_trigger.unique_key })
       ensure
-        R3x::RunWorkflowJob.singleton_class.send(:define_method, :perform_later, original_perform_later)
+        workflow_class.singleton_class.send(:define_method, :perform_later, original_perform_later)
       end
 
       assert_equal "enqueue failed", error.message
@@ -142,6 +140,7 @@ def run(ctx)
       end
 
       Workflow::Registry.register(workflow_class)
+      workflow_class
     end
   end
 end