Handle continuable workflow resumes in CLI

Author: zewelor

AGENTS.md

@@ -170,6 +170,7 @@ This is a Rails API app for the `r3x` Ruby-native workflow engine. Keep changes
 - Prefer expressive stdlib methods (`values_at`, `slice`, `filter_map`, `each_with_object`, `sum`, `tally`) over verbose manual loops when they make intent clearer.
 - For conditional hash keys, build a base hash, assign optional keys, and return it. Avoid `Hash#tap`/`merge` for simple conditional key assignment.
 - Method chaining is fine when it makes data flow clearer; introduce locals when the intermediate value is reused or has important meaning.
+- Prefer one-line `tap { it... }` for a single obvious post-construction tweak on the returned object. Use a named block parameter or multiline block when the object role is not obvious, there is more than one mutation, or the block contains branching/side effects that deserve a name.
 - Do not remove existing code comments without explicit user approval.
 
 ## Ruby Version Updates

bin/workflow

@@ -1,4 +1,6 @@
 #!/usr/bin/env ruby
+# frozen_string_literal: true
+
 ENV["R3X_RUNTIME_PROFILE"] = "workflow_cli"
 
 require_relative "../config/environment"
@@ -18,10 +20,11 @@ class WorkflowCli < Thor
   desc "run PATH", "Run a workflow from file path"
   option :dry_run, type: :boolean, desc: "Run with global dry run mode (clients skip external side effects)"
   option :skip_cache, type: :boolean, desc: "Run without workflow cache (with_cache blocks execute directly)"
+  option :skip_wait, type: :boolean, desc: "Skip waits between resumable workflow executions"
   option :verbose, type: :boolean, aliases: "-v", desc: "Enable verbose logging (debug level)"
   def execute(path)
     Rails.logger.level = Logger::DEBUG if options[:verbose]
-    cli.run(path, dry_run: options[:dry_run], skip_cache: options[:skip_cache])
+    cli.run(path, dry_run: options[:dry_run], skip_cache: options[:skip_cache], wait: !options[:skip_wait])
   end
 
   desc "list", "List available workflows"

docs/workflows.md

@@ -189,6 +189,7 @@ bin/workflow info <workflow_key>
 bin/workflow run workflows/<workflow_name>/workflow.rb
 bin/workflow run --dry-run workflows/<workflow_name>/workflow.rb
 bin/workflow run --skip-cache workflows/<workflow_name>/workflow.rb
+bin/workflow run --skip-wait workflows/<workflow_name>/workflow.rb
 ```
 
 For an included workflow in this checkout:
@@ -202,10 +203,14 @@ bin/workflow run --dry-run workflows/porto_santo_news/workflow.rb
 - `run` always takes a direct path to a `workflow.rb` file.
 - In `development` and `test`, `bin/workflow run` defaults to dry-run, so dry-run-aware clients avoid
   real side effects.
+- `run` resumes `ActiveJob::Continuable` interruptions in-process until the workflow completes.
+  For `isolated: true` steps it waits according to the workflow's `resume_options[:wait]`, matching
+  queued execution.
 - `--dry-run` explicitly enables dry-run for that run (`R3X_DRY_RUN=true`).
 - `--no-dry-run` explicitly disables dry-run for that run (`R3X_DRY_RUN=false`), even in
   `development`.
 - `--skip-cache` sets `R3X_SKIP_CACHE=true` for that run and bypasses `with_cache`.
+- `--skip-wait` fast-forwards through local Continuable waits when debugging isolated steps.
 - Use `--dry-run --skip-cache` together when you want a fresh, low-risk local run:
 
   ```bash

lib/r3x/workflow/cli.rb

@@ -9,10 +9,10 @@ def initialize(stdout: $stdout, pack_loader: PackLoader, registry: Registry)
         @registry = registry
       end
 
-      def run(path, dry_run: nil, skip_cache: false)
+      def run(path, dry_run: nil, skip_cache: false, wait: true)
         with_run_env(dry_run:, skip_cache:) do
           stdout.puts run_message(path, dry_run_explicit: dry_run == true)
-          load_workflow(path).new.perform
+          run_workflow(load_workflow(path), wait:)
         end
       end
 
@@ -47,6 +47,43 @@ def info(key)
 
       attr_reader :pack_loader, :registry, :stdout
 
+      def run_workflow(workflow_class, wait:)
+        workflow = workflow_class.new
+
+        loop do
+          return workflow.perform
+        rescue ActiveJob::Continuation::Interrupt
+          # Isolated Continuable steps interrupt so a queue adapter can persist progress
+          # and resume the job later. The local CLI runs in-process, so it carries that
+          # serialized continuation forward itself.
+          max_resumptions = workflow.class.max_resumptions
+
+          if max_resumptions && workflow.resumptions >= max_resumptions
+            raise ActiveJob::Continuation::ResumeLimitError, "Job was resumed a maximum of #{max_resumptions} times"
+          end
+
+          sleep resume_delay_for(workflow) if wait
+          workflow = workflow.class.deserialize(workflow.serialize).tap { it.resumptions += 1 }
+        end
+      end
+
+      def resume_delay_for(workflow)
+        resume_options = workflow.class.resume_options
+
+        if resume_options[:wait_until]
+          return [resume_options[:wait_until].to_f - Time.current.to_f, 0].max
+        end
+
+        return 0 unless resume_options[:wait]
+
+        workflow.send(
+          :determine_delay,
+          seconds_or_duration_or_algorithm: resume_options[:wait],
+          executions: workflow.resumptions + 1,
+          jitter: 0,
+        )
+      end
+
       def load_workflow(path)
         full_path = workflow_file_path(path)
         require full_path

test/fixtures/workflows/continuable_workflow/workflow.rb

@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+module Workflows
+  class ContinuableWorkflow < R3x::Workflow::Base
+    EVENTS = Concurrent::Array.new
+
+    self.resume_options = { wait: 2.minutes }
+
+    trigger :manual
+    on_complete { self.class.events << "complete" }
+
+    class << self
+      def events
+        EVENTS
+      end
+
+      def reset_events!
+        EVENTS.clear
+      end
+    end
+
+    def run
+      step :first do
+        self.class.events << "first"
+      end
+
+      step :second, isolated: true do
+        self.class.events << "second"
+      end
+
+      step :third, isolated: true do
+        self.class.events << "third"
+      end
+
+      { "events" => self.class.events.dup }
+    end
+  end
+end

test/integration/workflow_cli_test.rb

@@ -21,6 +21,12 @@ class WorkflowCliTest < ActiveSupport::TestCase
     assert_includes output, "Running with dry run: #{@fixture_path}"
   end
 
+  test "run command accepts skip wait option" do
+    output = run_cli("run --skip-wait #{@fixture_path}")
+
+    assert_includes output, "Running with dry run: #{@fixture_path}"
+  end
+
   test "nonexistent file shows error" do
     output = run_cli("run /nonexistent/path.rb", allow_failure: true)
 

test/lib/r3x/workflow/cli_test.rb

@@ -9,9 +9,11 @@ class CliTest < ActiveSupport::TestCase
       setup do
         @fixture_dir = Rails.root.join("test/fixtures/workflows")
         @fixture_path = @fixture_dir.join("test_workflow/workflow.rb")
+        @continuable_fixture_path = @fixture_dir.join("continuable_workflow/workflow.rb")
         @original_workflow_paths = ENV["R3X_WORKFLOW_PATHS"]
         ENV["R3X_WORKFLOW_PATHS"] = @fixture_dir.to_s
         R3x::Workflow::PackLoader.load!(rebuild_registry: true)
+        Workflows::ContinuableWorkflow.reset_events!
       end
 
       teardown do
@@ -65,6 +67,61 @@ def self.all = []
         assert_equal({ "test" => true, "message" => "Test workflow executed successfully" }, result)
       end
 
+      test "run resumes continuable workflows until completion" do
+        output = StringIO.new
+        cli = Cli.new(stdout: output)
+        cli.stubs(:sleep)
+
+        result = cli.run(@continuable_fixture_path.to_s)
+
+        assert_equal %w[first second third complete], Workflows::ContinuableWorkflow.events
+        assert_equal({ "events" => %w[first second third] }, result)
+      end
+
+      test "run waits between continuable workflow resumptions by default" do
+        output = StringIO.new
+        cli = Cli.new(stdout: output)
+        cli.expects(:sleep).twice.with(2.minutes.to_i)
+
+        cli.run(@continuable_fixture_path.to_s)
+      end
+
+      test "run supports polynomial wait between continuable workflow resumptions" do
+        original_resume_options = Workflows::ContinuableWorkflow.resume_options
+        Workflows::ContinuableWorkflow.resume_options = { wait: :polynomially_longer }
+        output = StringIO.new
+        cli = Cli.new(stdout: output)
+        sleeps = sequence("sleeps")
+        cli.expects(:sleep).with(3).in_sequence(sleeps)
+        cli.expects(:sleep).with(18).in_sequence(sleeps)
+
+        cli.run(@continuable_fixture_path.to_s)
+      ensure
+        Workflows::ContinuableWorkflow.resume_options = original_resume_options
+      end
+
+      test "run supports wait until between continuable workflow resumptions" do
+        original_resume_options = Workflows::ContinuableWorkflow.resume_options
+        travel_to Time.zone.local(2026, 1, 1, 12) do
+          Workflows::ContinuableWorkflow.resume_options = { wait_until: 30.seconds.from_now }
+          output = StringIO.new
+          cli = Cli.new(stdout: output)
+          cli.expects(:sleep).twice.with(30.0)
+
+          cli.run(@continuable_fixture_path.to_s)
+        end
+      ensure
+        Workflows::ContinuableWorkflow.resume_options = original_resume_options
+      end
+
+      test "run can skip waiting between continuable workflow resumptions" do
+        output = StringIO.new
+        cli = Cli.new(stdout: output)
+        cli.expects(:sleep).never
+
+        cli.run(@continuable_fixture_path.to_s, wait: false)
+      end
+
       test "run supports dry run and skip cache messaging without leaking env overrides" do
         output = StringIO.new