[DO NOT MERGE] Implement operator commands for Standalone Activities

CHANGELOG.md

@@ -21,6 +21,11 @@ to docs, or any other relevant information.
 
 ### Added
 
+#### Standalone Activity operator commands
+
+- `Client::ActivityHandle` now supports operator commands for standalone activities: `#pause`,
+  `#unpause`, `#reset`, and `#update_options`.
+
 ### Changed
 
 ### Deprecated

temporalio/lib/temporalio/client/activity_execution.rb

@@ -46,6 +46,12 @@ def schedule_time
         Internal::ProtoUtils.timestamp_to_time(@raw_info.schedule_time)
       end
 
+      # @return [Time, nil] When the first activity task was made available for dispatch. Equals
+      #   schedule_time + start_delay; equal to schedule_time when no start delay is set.
+      def execution_time
+        Internal::ProtoUtils.timestamp_to_time(@raw_info.execution_time)
+      end
+
       # @return [Time, nil] When the activity reached a terminal state.
       def close_time
         Internal::ProtoUtils.timestamp_to_time(@raw_info.close_time)
@@ -112,7 +118,17 @@ def heartbeat_timeout
           Internal::ProtoUtils.duration_to_seconds(@raw_info.heartbeat_timeout)
         end
 
-        # @return [Boolean] Whether the activity has recorded any heartbeat details.
+        # @return [Float, nil] Delay in seconds before the first activity task is made available for
+        #   dispatch. Not applied to retry attempts.
+        def start_delay
+          Internal::ProtoUtils.duration_to_seconds(@raw_info.start_delay)
+        end
+
+        # Whether heartbeat details are present on this description. False when the activity
+        # recorded none, and also when {ActivityHandle#describe} was called without
+        # `include_heartbeat_details:`.
+        #
+        # @return [Boolean] Whether heartbeat details are present.
         def has_heartbeat_details? # rubocop:disable Naming/PredicatePrefix
           !@raw_info.heartbeat_details&.payloads.nil? && !@raw_info.heartbeat_details.payloads.empty?
         end
@@ -125,6 +141,57 @@ def heartbeat_details(hints: nil)
           @data_converter.from_payloads(@raw_info.heartbeat_details, hints:)
         end
 
+        # Whether the activity's input is present. False unless {ActivityHandle#describe} was
+        # called with `include_input:`.
+        #
+        # @return [Boolean] Whether input is present.
+        def has_input? # rubocop:disable Naming/PredicatePrefix
+          !@raw_description.input.nil?
+        end
+
+        # Deserialized activity input, one element per argument. Empty when no input is present.
+        #
+        # @param hints [Array<Object>, nil] Hints, if any, to assist conversion.
+        # @return [Array<Object>] Converted arguments.
+        def input(hints: nil)
+          @data_converter.from_payloads(@raw_description.input, hints:)
+        end
+
+        # Whether the activity closed with a successful result. False while the activity is still
+        # running, when it closed with a failure, and when {ActivityHandle#describe} was called
+        # without `include_outcome:`.
+        #
+        # @return [Boolean] Whether a result is present.
+        def has_result? # rubocop:disable Naming/PredicatePrefix
+          @raw_description.outcome&.value == :result
+        end
+
+        # Deserialized result the activity closed with. Nil when no result is present (still
+        # running, closed with a failure, or `include_outcome:` was not requested).
+        #
+        # @param result_hint [Object, nil] Hint, if any, to assist conversion.
+        # @return [Object, nil] Converted result.
+        def result(result_hint: nil)
+          return nil unless has_result?
+
+          @data_converter.from_payloads(
+            @raw_description.outcome.result, hints: Array(result_hint)
+          ).first
+        end
+
+        # Failure the activity closed with. Nil when the activity did not close with a failure or
+        # when {ActivityHandle#describe} was called without `include_outcome:`.
+        #
+        # This is the terminal outcome; {#last_failure} is the failure of the most recent attempt,
+        # which may be set while the activity is still retrying.
+        #
+        # @return [Error::Failure, nil] Converted failure.
+        def failure
+          return nil unless @raw_description.outcome&.value == :failure
+
+          @data_converter.from_failure(@raw_description.outcome.failure)
+        end
+
         # @return [RetryPolicy] Retry policy in effect for this activity.
         def retry_policy
           RetryPolicy._from_proto(@raw_info.retry_policy)
@@ -145,6 +212,15 @@ def attempt
           @raw_info.attempt
         end
 
+        # Whether a last failure is present on this description. False when the activity has no
+        # failed attempt, and also when {ActivityHandle#describe} was called without
+        # `include_last_failure:`.
+        #
+        # @return [Boolean] Whether a last failure is present.
+        def has_last_failure? # rubocop:disable Naming/PredicatePrefix
+          !@raw_info.last_failure.nil?
+        end
+
         # @return [Error::Failure, nil] Failure of the last failed attempt if any.
         def last_failure
           return nil unless @raw_info.last_failure

temporalio/lib/temporalio/client/activity_execution_options.rb

@@ -0,0 +1,39 @@
+# frozen_string_literal: true
+
+require 'temporalio/internal/proto_utils'
+require 'temporalio/priority'
+require 'temporalio/retry_policy'
+
+module Temporalio
+  class Client
+    # The resolved options of a standalone activity execution, as returned by
+    # {ActivityHandle#update_options}. Reflects the activity's options as the server resolved them
+    # after the update was applied.
+    #
+    # WARNING: Standalone Activities are experimental.
+    ActivityExecutionOptions = Data.define(
+      :task_queue,
+      :schedule_to_close_timeout,
+      :schedule_to_start_timeout,
+      :start_to_close_timeout,
+      :heartbeat_timeout,
+      :retry_policy,
+      :priority,
+      :start_delay
+    ) do
+      # @!visibility private
+      def self._from_proto(options)
+        new(
+          task_queue: Internal::ProtoUtils.string_or(options.task_queue&.name, nil),
+          schedule_to_close_timeout: Internal::ProtoUtils.duration_to_seconds(options.schedule_to_close_timeout),
+          schedule_to_start_timeout: Internal::ProtoUtils.duration_to_seconds(options.schedule_to_start_timeout),
+          start_to_close_timeout: Internal::ProtoUtils.duration_to_seconds(options.start_to_close_timeout),
+          heartbeat_timeout: Internal::ProtoUtils.duration_to_seconds(options.heartbeat_timeout),
+          retry_policy: options.retry_policy ? RetryPolicy._from_proto(options.retry_policy) : nil,
+          priority: Priority._from_proto(options.priority),
+          start_delay: Internal::ProtoUtils.duration_to_seconds(options.start_delay)
+        )
+      end
+    end
+  end
+end

temporalio/lib/temporalio/client/activity_execution_status.rb

@@ -15,6 +15,7 @@ module ActivityExecutionStatus
       CANCELED = Api::Enums::V1::ActivityExecutionStatus::ACTIVITY_EXECUTION_STATUS_CANCELED
       TERMINATED = Api::Enums::V1::ActivityExecutionStatus::ACTIVITY_EXECUTION_STATUS_TERMINATED
       TIMED_OUT = Api::Enums::V1::ActivityExecutionStatus::ACTIVITY_EXECUTION_STATUS_TIMED_OUT
+      PAUSED = Api::Enums::V1::ActivityExecutionStatus::ACTIVITY_EXECUTION_STATUS_PAUSED
     end
   end
 end

temporalio/lib/temporalio/client/activity_handle.rb

@@ -2,8 +2,12 @@
 
 require 'temporalio/api'
 require 'temporalio/client/activity_execution'
+require 'temporalio/client/activity_execution_options'
 require 'temporalio/client/interceptor'
 require 'temporalio/error'
+require 'temporalio/internal/proto_utils'
+require 'temporalio/priority'
+require 'temporalio/retry_policy'
 
 module Temporalio
   class Client
@@ -12,6 +16,17 @@ class Client
     #
     # WARNING: Standalone Activities are experimental.
     class ActivityHandle
+      UPDATABLE_OPTION_PATHS = {
+        task_queue: 'task_queue.name',
+        schedule_to_close_timeout: 'schedule_to_close_timeout',
+        schedule_to_start_timeout: 'schedule_to_start_timeout',
+        start_to_close_timeout: 'start_to_close_timeout',
+        heartbeat_timeout: 'heartbeat_timeout',
+        retry_policy: 'retry_policy',
+        priority: 'priority',
+        start_delay: 'start_delay'
+      }.freeze
+
       # @return [String] ID for the activity.
       attr_reader :id
 
@@ -55,15 +70,35 @@ def result(result_hint: nil, rpc_options: nil)
 
       # Describe the activity.
       #
+      # The payload-bearing fields are opt-in because they can be arbitrarily large; request them
+      # only when needed. Each has a corresponding predicate on the returned description that
+      # reports whether the server supplied it.
+      #
+      # @param include_input [Boolean] If true and the activity received input, include the input.
+      # @param include_outcome [Boolean] If true and the activity is closed, include the outcome.
+      # @param include_heartbeat_details [Boolean] If true and the activity recorded heartbeat
+      #   details, include them.
+      # @param include_last_failure [Boolean] If true and the activity has a failed attempt, include
+      #   the last failure.
       # @param rpc_options [RPCOptions, nil] Advanced RPC options.
       #
       # @return [ActivityExecution::Description] Activity description.
       # @raise [Error::RPCError] RPC error from call.
-      def describe(rpc_options: nil)
+      def describe(
+        include_input: false,
+        include_outcome: false,
+        include_heartbeat_details: false,
+        include_last_failure: false,
+        rpc_options: nil
+      )
         @client._impl.describe_activity(
           Interceptor::DescribeActivityInput.new(
             activity_id: id,
             activity_run_id: run_id,
+            include_input:,
+            include_outcome:,
+            include_heartbeat_details:,
+            include_last_failure:,
             rpc_options:
           )
         )
@@ -103,6 +138,142 @@ def terminate(reason = nil, rpc_options: nil)
         nil
       end
 
+      # Pause the activity. A paused activity is not scheduled or retried until it is unpaused via
+      # {#unpause}.
+      #
+      # WARNING: Standalone Activities are experimental.
+      #
+      # @param reason [String, nil] Optional reason recorded on the server.
+      # @param rpc_options [RPCOptions, nil] Advanced RPC options.
+      # @raise [Error::RPCError] RPC error from call.
+      def pause(reason = nil, rpc_options: nil)
+        @client._impl.pause_activity(
+          Interceptor::PauseActivityInput.new(
+            activity_id: id,
+            activity_run_id: run_id,
+            reason:,
+            rpc_options:
+          )
+        )
+        nil
+      end
+
+      # Unpause the activity, allowing it to be scheduled or retried again.
+      #
+      # WARNING: Standalone Activities are experimental.
+      #
+      # @param reason [String, nil] Optional reason recorded on the server.
+      # @param jitter [Float, nil] If set, the activity will start at a random time within this
+      #   duration (in seconds).
+      # @param rpc_options [RPCOptions, nil] Advanced RPC options.
+      # @raise [Error::RPCError] RPC error from call.
+      def unpause(reason: nil, jitter: nil, rpc_options: nil)
+        @client._impl.unpause_activity(
+          Interceptor::UnpauseActivityInput.new(
+            activity_id: id,
+            activity_run_id: run_id,
+            reason:,
+            jitter:,
+            rpc_options:
+          )
+        )
+        nil
+      end
+
+      # Reset the activity. Resetting sets the attempt count back to the start, resets the activity's
+      # timeouts, and clears any recorded heartbeat details.
+      #
+      # WARNING: Standalone Activities are experimental.
+      #
+      # @param keep_paused [Boolean] If true and the activity is paused, it remains paused after reset.
+      # @param jitter [Float, nil] If set and the activity is in backoff, it will start at a random
+      #   time within this duration (in seconds).
+      # @param restore_original_options [Boolean] If true, restore the activity options to the
+      #   originals it was created with.
+      # @param reset_heartbeat [Boolean] If true, additionally discard any persisted heartbeat details.
+      # @param rpc_options [RPCOptions, nil] Advanced RPC options.
+      # @raise [Error::RPCError] RPC error from call.
+      def reset(keep_paused: false, jitter: nil, restore_original_options: false,
+                reset_heartbeat: false, rpc_options: nil)
+        @client._impl.reset_activity(
+          Interceptor::ResetActivityInput.new(
+            activity_id: id,
+            activity_run_id: run_id,
+            keep_paused:,
+            jitter:,
+            restore_original_options:,
+            reset_heartbeat:,
+            rpc_options:
+          )
+        )
+        nil
+      end
+
+      # Update the activity's options. Only the options you actually pass are changed; anything you
+      # omit is left as-is. Passing an option explicitly as `nil` clears it.
+      #
+      # WARNING: Standalone Activities are experimental.
+      #
+      # @param restore_original [Boolean] If true, restore the options to the originals the activity
+      #   was created with. Mutually exclusive with any other option.
+      # @param rpc_options [RPCOptions, nil] Advanced RPC options.
+      # @param options [Hash{Symbol => Object, nil}] The options to change. Keys must be drawn from
+      #   {UPDATABLE_OPTION_PATHS}; anything else raises `ArgumentError`.
+      # @option options [String, Symbol, nil] :task_queue New task queue.
+      # @option options [Float, nil] :schedule_to_close_timeout New schedule-to-close timeout in seconds.
+      # @option options [Float, nil] :schedule_to_start_timeout New schedule-to-start timeout in seconds.
+      # @option options [Float, nil] :start_to_close_timeout New start-to-close timeout in seconds.
+      # @option options [Float, nil] :heartbeat_timeout New heartbeat timeout in seconds.
+      # @option options [RetryPolicy, nil] :retry_policy New retry policy.
+      # @option options [Priority, nil] :priority New priority.
+      # @option options [Float, nil] :start_delay New start delay in seconds.
+      #
+      # @return [ActivityExecutionOptions] The activity options after the update.
+      #
+      # @raise [ArgumentError] If an unknown option is given, if `restore_original` is combined with
+      #   any other option, or if no option is provided and `restore_original` is false.
+      # @raise [Error::RPCError] RPC error from call.
+      def update_options(restore_original: false, rpc_options: nil, **options)
+        unknown = options.keys - UPDATABLE_OPTION_PATHS.keys
+        unless unknown.empty?
+          raise ArgumentError,
+                "Unknown option(s): #{unknown.join(', ')}. " \
+                "Expected any of: #{UPDATABLE_OPTION_PATHS.keys.join(', ')}"
+        end
+
+        if restore_original && !options.empty?
+          raise ArgumentError, 'restore_original cannot be combined with any other option'
+        elsif !restore_original && options.empty?
+          raise ArgumentError, 'At least one option must be set, or restore_original must be used'
+        end
+
+        proto = Api::Activity::V1::ActivityOptions.new
+        if options.key?(:task_queue) && (task_queue = options[:task_queue])
+          proto.task_queue = Api::TaskQueue::V1::TaskQueue.new(name: task_queue.to_s)
+        end
+        %i[schedule_to_close_timeout schedule_to_start_timeout start_to_close_timeout
+           heartbeat_timeout start_delay].each do |name|
+          next unless options.key?(name)
+
+          proto[name.to_s] = Internal::ProtoUtils.seconds_to_duration(options[name])
+        end
+        proto.retry_policy = options[:retry_policy]&._to_proto if options.key?(:retry_policy)
+        proto.priority = options[:priority]&._to_proto if options.key?(:priority)
+
+        @client._impl.update_activity_options(
+          Interceptor::UpdateActivityOptionsInput.new(
+            activity_id: id,
+            activity_run_id: run_id,
+            activity_options: proto,
+            update_mask: Google::Protobuf::FieldMask.new(
+              paths: options.keys.map { |k| UPDATABLE_OPTION_PATHS.fetch(k) }
+            ),
+            restore_original:,
+            rpc_options:
+          )
+        )
+      end
+
       private
 
       def _process_outcome(outcome, hint)