Fail a run as lost when its worker stops

CHANGELOG.md

@@ -5,6 +5,33 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+Upgrading the dashboard: run `bin/rails generate action_agent:install --skip`
+and `bin/rails db:migrate` to install migration 033,
+`add_agent_run_execution_leases`. Until it runs, agent runs take no leases
+and a run whose worker dies stays running as before.
+
+### Fixed
+
+- **A run whose worker died is failed as lost instead of running forever**
+  (`actionagent`). Since a run is claimed once, a worker that died after
+  its claim left the run `running` for good: a redelivered job leaves a
+  running run alone. Each claim (the execution job's, the resume job's,
+  `Agent#test_execute`'s) now takes a lease (migration 033,
+  `config.run_lease_duration`, 30 minutes by default) that each LLM or
+  tool call renews. A running run whose lease ran out is failed with
+  `error_class` `ActionAgent::AgentRun::WorkerLost` when its page or the
+  run list is read, or when `ActionAgent::AgentRunExpiryJob` runs (not
+  scheduled by default). Nothing is executed again: the run's provider and
+  tool calls may already have taken effect. A worker that was only slow
+  has its result discarded, as a cancelled run's is, and a result now
+  only lands while the run holds its worker's lease, so whatever ended
+  the run first is kept. `test_execute` records its result the same way,
+  so a cancel during a synchronous test run is no longer overwritten. A
+  run with no lease, such as one running when the migration is installed,
+  is never failed as lost.
+
 ## [1.9.8] - 2026-11-12
 
 The dashboard holds a run to its catalog's pass bars: a product's and a

actionagent/app/controllers/action_agent/api/agent_runs_controller.rb

@@ -8,6 +8,7 @@ class AgentRunsController < BaseController
       # GET /api/runs/:id
       def show
         @run.expire_overdue_input_requests!
+        @run.expire_if_lost!
         render json: {
           run: run_json(@run),
           messages: interaction_messages(@run),
@@ -29,6 +30,8 @@ def cancel
       def index
         # Scoped to the caller's own agents: this listed (and counted) every
         # run in the database regardless of who owned it.
+        # A run whose worker was lost is listed as failed, not as running.
+        AgentRun.expire_lost!(AgentRun.where(agent: owner_agents))
         scope = AgentRun.includes(:agent, :pending_input_requests).with_attachments.where(agent: owner_agents).recent
 
         scope = scope.where(agent_id: params[:agent_id]) if params[:agent_id].present?

actionagent/app/jobs/action_agent/agent_execution_job.rb

@@ -12,6 +12,9 @@ def perform(run_id)
       # retry of a failed run would re-run the generation.
       return unless claim_pending(run)
 
+      # This worker's lease: only while the run is still running under it is
+      # the result recorded.
+      token = run.execution_token
       run.add_log("Starting execution", level: :info)
 
       begin
@@ -20,12 +23,12 @@ def perform(run_id)
         # Build the agent class dynamically based on configuration
         result = execute_agent(agent_record, run)
 
-        run.unless_cancelled do
+        run.unless_cancelled(token: token) do
           run.record_result!(result, segment_started_at: run.started_at)
           run.add_log(result[:input_requests].present? ? "Execution paused for input" : "Execution completed successfully", level: :info)
         end
       rescue => e
-        run.unless_cancelled do
+        run.unless_cancelled(token: token) do
           run.record_failure!(e)
           run.add_log("Execution failed: #{e.message}", level: :error)
         end
@@ -46,11 +49,12 @@ def perform(run_id)
     # with Rails 7.2, whose deferred transactions cannot write once another
     # writer has committed. Nothing is held over the generation, so a cancel
     # never waits for it. The update skips callbacks, so the status change is
-    # broadcast here.
+    # broadcast here. The claim takes the run's lease (AgentRun.lease_attributes),
+    # so the reloaded run carries this worker's token.
     def claim_pending(run)
       now = Time.current
       claimed = AgentRun.where(id: run.id, status: AgentRun.statuses[:pending])
-        .update_all(status: AgentRun.statuses[:running], started_at: now, updated_at: now)
+        .update_all(status: AgentRun.statuses[:running], started_at: now, updated_at: now, **AgentRun.lease_attributes(now))
       return false if claimed.zero?
 
       run.reload

actionagent/app/jobs/action_agent/agent_resume_job.rb

@@ -24,6 +24,8 @@ def perform(input_request_id)
       pause = claim_pause(run, request)
       return unless pause
 
+      token = run.execution_token
+
       segment_started_at = Time.current
       secrets = pause.select { |paused| paused.kind == "secret" }.filter_map(&:answer)
 
@@ -39,13 +41,13 @@ def perform(input_request_id)
           }
         )
 
-        run.unless_cancelled do
+        run.unless_cancelled(token: token) do
           run.record_result!(result, segment_started_at: segment_started_at)
           run.add_log(result[:input_requests].present? ? "Execution paused for input" : "Execution completed successfully", level: :info)
         end
       rescue => e
         message = ActiveAgent::InputRequest.scrub(e.message, secrets)
-        run.unless_cancelled do
+        run.unless_cancelled(token: token) do
           run.record_failure!(e, message: message)
           run.add_log("Execution failed: #{message}", level: :error)
         end
@@ -60,15 +62,16 @@ def perform(input_request_id)
     # Moves the run from awaiting_input to running when every request of the
     # pause is settled, under the run's row lock, and returns the pause's
     # requests in call order; nil when this job is not the one to resume it,
-    # or when the request's pause is not the one the run waits on.
+    # or when the request's pause is not the one the run waits on. The resumed
+    # segment takes a lease of its own, under a new token.
     def claim_pause(run, request)
       run.with_lock do
         next nil unless run.awaiting_input? && request.current_pause?
 
         pause = request.pause_requests.order(:id).to_a
         next nil unless pause.all? { |paused| paused.answered? || paused.declined? }
 
-        run.update!(status: :running)
+        run.update!(status: :running, **AgentRun.lease_attributes)
         pause
       end
     end

actionagent/app/jobs/action_agent/agent_run_expiry_job.rb

@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+module ActionAgent
+  # Fails the runs whose lease ran out (AgentRun.expire_lost!): running, with
+  # no claim, renewal or result from their worker within
+  # ActionAgent.run_lease_duration. A run's own page and the run list fail
+  # the lost runs they reach; this job reaches the rest. It never executes a
+  # run again.
+  #
+  # Not scheduled by default. To schedule it:
+  #
+  #   # config/recurring.yml
+  #   agent_run_expiry:
+  #     class: ActionAgent::AgentRunExpiryJob
+  #     schedule: every 5 minutes
+  class AgentRunExpiryJob < ApplicationJob
+    queue_as :default
+
+    def perform
+      AgentRun.expire_lost!
+    end
+  end
+end

actionagent/app/models/action_agent/agent.rb

@@ -492,22 +492,26 @@ def execute(input_prompt, action: nil, attachments: [], actor: nil, runtime_sand
 
     # Quick test execution (synchronous). A run that pauses for input comes
     # back `awaiting_input` with its requests stored, and AgentResumeJob
-    # finishes it once they are answered.
+    # finishes it once they are answered. The run is created running, under
+    # a lease, as a job's claim would leave it, and its result is recorded
+    # as a job's is: not over a cancel, nor once the run left its lease.
     def test_execute(input_prompt, action: nil, attachments: [], actor: nil, runtime_sandbox: nil, **params)
       ensure_executable!
       run = create_run(
         input_prompt, action: action, attachments: attachments, params: params,
-        actor: actor, runtime_sandbox: runtime_sandbox, status: :running, started_at: Time.current
+        actor: actor, runtime_sandbox: runtime_sandbox, status: :running, started_at: Time.current,
+        **AgentRun.lease_attributes
       )
       run.actor = actor
+      token = run.execution_token
 
       begin
         # Build and execute the agent
         result = AgentExecutionService.call(self, run)
 
-        run.record_result!(result, segment_started_at: run.started_at)
+        run.unless_cancelled(token: token) { run.record_result!(result, segment_started_at: run.started_at) }
       rescue => e
-        run.record_failure!(e)
+        run.unless_cancelled(token: token) { run.record_failure!(e) }
       end
 
       run

actionagent/app/models/action_agent/agent_run.rb

@@ -19,6 +19,11 @@ def initialize(message = "Attachments need Active Storage in the host app (run `
       end
     end
 
+    # The error class recorded on a run failed as lost: running past its
+    # lease with no news from its worker (see #expire_if_lost!). Never
+    # raised; it names the failure, as record_failure! names any other.
+    class WorkerLost < StandardError; end
+
     # Files uploaded with the run. The execution service delivers them to
     # the model (images and PDFs as data URIs, text inlined) and the
     # persisted user message keeps a manifest of them.
@@ -165,6 +170,37 @@ def self.input_param_sql(key)
       end
     end
 
+    # Whether migration 033's lease columns are installed. A host that
+    # upgraded the gem without running the generator takes no leases and
+    # runs as before, instead of failing every claim on a missing column.
+    def self.execution_leases?
+      (%w[execution_token execution_expires_at] - column_names).empty?
+    end
+
+    # The attributes a claim writes to take a run for one execution segment:
+    # a token of its own, and an expiry ActionAgent.run_lease_duration from
+    # +now+. Empty when leases are off (no columns, or no duration).
+    # @return [Hash]
+    def self.lease_attributes(now = Time.current)
+      duration = ActionAgent.run_lease_duration
+      return {} unless duration && execution_leases?
+
+      { execution_token: SecureRandom.uuid, execution_expires_at: now + duration }
+    end
+
+    # Pushes the lease's expiry out by ActionAgent.run_lease_duration, while
+    # the run is still running under this copy's token. A run that finished,
+    # paused, or was taken under another token is left alone. Returns
+    # whether the lease was renewed.
+    # @return [Boolean]
+    def renew_lease!(now = Time.current)
+      duration = ActionAgent.run_lease_duration
+      return false unless duration && self.class.execution_leases? && execution_token
+
+      self.class.where(id: id, status: self.class.statuses[:running], execution_token: execution_token)
+        .update_all(execution_expires_at: now + duration) == 1
+    end
+
     # Callbacks
     before_validation :set_trace_id, on: :create
     after_update_commit :broadcast_update, if: :saved_change_to_status?
@@ -185,6 +221,7 @@ def add_log(message, level: :info)
     # a "started" event is pending until a "done"/"error" with the same eid
     # lands. update_column: no validations/callbacks, safe from the run's own
     # execution thread; reads current DB state so add_log interleaves safely.
+    # An event is the run's sign of progress, so it renews the run's lease.
     def append_event(eid:, kind:, label:, status: "done", detail: nil, duration_ms: nil)
       event = {
         "at" => Time.current.iso8601(3),
@@ -197,6 +234,7 @@ def append_event(eid:, kind:, label:, status: "done", detail: nil, duration_ms:
       event["duration_ms"] = duration_ms if duration_ms
       current = self.class.where(id: id).pick(:logs) || []
       update_column(:logs, current + [ event ])
+      renew_lease!
       event
     end
 
@@ -346,6 +384,54 @@ def cancel!(reason = "Cancelled by user")
       cancelled
     end
 
+    # Running runs whose lease ran out: no claim, renewal or result from
+    # their worker within ActionAgent.run_lease_duration. A run without a
+    # lease (claimed before migration 033, or with leases off) is never
+    # among them. Needs the lease columns (see .execution_leases?).
+    scope :lost, ->(now = Time.current) {
+      running.where.not(execution_token: nil).where(execution_expires_at: ..now)
+    }
+
+    # Fails every lost run in +relation+ (see #expire_if_lost!) and returns
+    # how many it failed. Nothing without the lease columns.
+    # @return [Integer]
+    def self.expire_lost!(relation = all, now: Time.current)
+      return 0 unless execution_leases?
+
+      relation.lost(now).find_each.count { |run| run.expire_if_lost!(now) }
+    end
+
+    # Fails the run as lost when it is still running under the lease this
+    # copy read and that lease has run out, in one conditional update, so a
+    # renewal, a result or a cancel that lands first wins and this changes
+    # nothing. The run is not executed again: its worker may have died, or
+    # may only be slow, and whatever its provider and tool calls did may
+    # already have happened; a slow worker's result is then discarded (see
+    # #unless_cancelled). Returns whether this call failed the run, which is
+    # reloaded either way it touches the database.
+    # @return [Boolean]
+    def expire_if_lost!(now = Time.current)
+      return false unless self.class.execution_leases? && running? && execution_token &&
+        execution_expires_at && execution_expires_at <= now
+
+      message = "The run's worker stopped reporting progress (its lease expired at #{execution_expires_at.iso8601}). " \
+                "The run was not executed again; its provider and tool calls may already have taken effect."
+      changed = self.class
+        .where(id: id, status: self.class.statuses[:running], execution_token: execution_token)
+        .where(execution_expires_at: ..now)
+        .update_all(
+          status: self.class.statuses[:failed], completed_at: now, error_message: message,
+          output_metadata: output_metadata.to_h.merge("error_class" => WorkerLost.name),
+          execution_token: nil, execution_expires_at: nil, updated_at: now
+        )
+      reload
+      return false if changed.zero?
+
+      add_log(message, level: :error)
+      broadcast_update
+      true
+    end
+
     # Expires the run's overdue requests for input, which fails the run (see
     # InputRequest#expire!). Returns the run, reloaded when a request
     # expired.
@@ -376,9 +462,9 @@ def record_result!(result, segment_started_at:)
       transaction do
         if result[:input_requests].present?
           InputRequest.record_pause!(self, result[:input_requests], checkpoint: result[:checkpoint])
-          update!(attributes.merge(status: :awaiting_input))
+          update!(attributes.merge(status: :awaiting_input, **lease_release_attributes))
         else
-          update!(attributes.merge(output: result[:output], status: :complete, completed_at: Time.current))
+          update!(attributes.merge(output: result[:output], status: :complete, completed_at: Time.current, **lease_release_attributes))
         end
       end
     end
@@ -395,19 +481,31 @@ def record_failure!(error, message: error.message)
         completed_at: Time.current,
         error_message: message,
         error_backtrace: error.backtrace&.first(10)&.join("\n"),
-        output_metadata: output_metadata.to_h.merge("error_class" => error.class.name)
+        output_metadata: output_metadata.to_h.merge("error_class" => error.class.name),
+        **lease_release_attributes
       )
     end
 
     # Yields under the run's row lock unless the run was cancelled meanwhile,
     # so a cancel that arrived while the run executed is never overwritten by
-    # the result. Returns whether it yielded.
-    def unless_cancelled
+    # the result. Given the +token+ of the worker's lease, it also yields only
+    # while the run still holds that token. Every write that ends a segment
+    # releases the lease, so a worker whose run was ended under it (failed as
+    # lost, say) or taken for another segment has its result discarded.
+    # Returns whether it yielded.
+    #
+    # @param token [String, nil] the lease token the worker's claim took; nil
+    #   (no lease) checks for a cancel only
+    def unless_cancelled(token: nil)
       with_lock do
         if cancelled?
           add_log("Execution finished after cancellation; result discarded", level: :info)
           next false
         end
+        if token && execution_token != token
+          add_log("Execution finished after its run was ended or taken by another worker; result discarded", level: :warn)
+          next false
+        end
 
         yield
         true
@@ -416,6 +514,12 @@ def unless_cancelled
 
     private
 
+    # Ending a segment ends its lease: nothing renews it, and no worker holds
+    # the run until the next claim.
+    def lease_release_attributes
+      self.class.execution_leases? ? { execution_token: nil, execution_expires_at: nil } : {}
+    end
+
     def add_tokens(recorded, segment)
       return recorded if segment.nil?
 

actionagent/lib/action_agent.rb

@@ -829,6 +829,18 @@ def claude_code_auth=(value)
     # @return [ActiveSupport::Duration, nil]
     attr_accessor :input_request_ttl
 
+    # How long a claimed run stays its worker's without news from it. Each
+    # claim (the execution job's, the resume job's, a synchronous test run's)
+    # takes a lease this long under a token of its own, and each LLM or tool
+    # call the run makes renews it. A run whose lease ran out is failed as
+    # lost once its page, the run list or AgentRunExpiryJob reaches it, and a
+    # worker that finishes after that has its result discarded. Nothing is
+    # executed again. 30 minutes by default: one provider call longer than
+    # this, with no progress in between, is taken for a lost worker. nil
+    # takes no leases, as before migration 033.
+    # @return [ActiveSupport::Duration, nil]
+    attr_accessor :run_lease_duration
+
     # Whether API keys and provider credentials are encrypted at rest with
     # Active Record Encryption. On by default, which requires the host app
     # to have run `rails db:encryption:init`. Turning it off stores those
@@ -1356,6 +1368,7 @@ def reset!
       @agent_polymorphic_name = nil
       @encrypt_credentials = true
       @input_request_ttl = 1.day
+      @run_lease_duration = 30.minutes
       @github_client_id = nil
       @github_client_secret = nil
       @github_oauth_scopes = "repo read:user"