feat(settings): super-admin background jobs console (#2682)

* feat(settings): super-admin background jobs console

Neither managed nor self-hosted production deployments have any view
into background job state (/sidekiq is only mounted outside
production), so a stuck sync, import, or export is invisible until a
user complains — and even then there is no way to act on it.

Adds /settings/background_jobs, gated by the same super-admin
Admin::BaseController as /settings/debug:

- Worker status header: Sidekiq processes, busy count, queue depths
  and latency, retry/scheduled/dead set sizes. Read through a
  fail-closed PORO (BackgroundJobConsole) — when Redis is unreachable
  the console says so and disables actions instead of pretending
  health (deliberate contrast to BackgroundJobHealth's fail-open).
- In-flight operations across all families: incomplete Syncs, Imports
  importing/reverting, ImportSessions importing, FamilyExports
  pending/processing, each with a liveness verdict derived from
  Sidekiq::Workers (job GlobalIDs in running payloads).
- One mutation: mark a presumed-lost operation as such (Sync → stale,
  Import → failed/revert_failed, PdfImport → claim released back to
  pending, ImportSession/FamilyExport → failed).

Guard rails on the mutation, server-side re-checked:

- refused while Redis state is unknown (fail closed)
- refused while the record's job is visibly executing
- refused until the record has been idle past 30 minutes, so a merely
  queued job cannot be shot down and then still run
- refused for parent Syncs with children still in flight (a parent
  legitimately has no live job of its own while children run)
- applied inside with_lock with a status re-check, so a job finishing
  between render and click wins
- audited as a DebugLogEntry (actor, prior status, family)

The cancel endpoint gets its own Rack Attack throttle since the
console deliberately lives under /settings rather than the throttled
/admin prefix (its 10 req/min limit would fight the page's polling).

* fix(jobs-console): count waiting jobs as live and harden cancel paths

Review feedback on #2682 (Codex, CodeRabbit):

- Liveness now covers jobs sitting in queues, the retry set, and the
  scheduled set, not just visibly-executing workers — a job waiting out
  a backlog or retry backoff WILL run later, and most affected job
  classes don't abort on a flipped status, so cancelling invited
  duplicate work. The backlog scan is bounded (5k entries); a truncated
  scan fails closed like redis_error?. The liveness column shows
  "Queued" for these
- find_record! resolves STI subclass names (TransactionImport,
  PdfImport, …) against a base-class whitelist via safe_constantize
  instead of a fixed name map, so non-UI callers naming the subclass
  don't 404
- A PdfImport stuck in reverting goes to revert_failed like every other
  import instead of being released to pending — pending presented a
  possibly half-reverted import as publishable again; only the AI
  extraction claim (importing) is released to pending
- Redis-unreachable warning renders via DS::Alert; operation id cast
  to_s before splitting; admin-cancel error copy moved to i18n

* fix(jobs-console): re-check the stuck window inside the cancel row lock

CodeRabbit round-2: cancellable? evaluates Sidekiq liveness outside the
with_lock transaction (re-running Redis calls under a row lock would be
worse), so a worker picking the job up between the liveness check and
the lock acquisition was invisible. Repeating the updated_at staleness
check inside the lock closes that window — a freshly-started job
touches the record, and the re-check refuses the cancel.

* fix(jobs-console): resolve record_type without reflection, i18n nits

Brakeman flagged safe_constantize on params[:record_type] as a
High-confidence UnsafeReflection (ci/scan_ruby). Replace the
constant lookup with a reverse lookup: find the id in each
cancellable base table and require the claimed type to match the
found record's class or its base class. STI subclass names still
resolve; unknown types still 404.

Also from DS Drift Patrol:
- "Sync · " label in _operation.html.erb now goes through
  t(".sync_label", type:)
- drop the redundant default: on background_jobs_label now that
  the key exists in the locale file

app/controllers/settings/background_jobs_controller.rb

@@ -0,0 +1,123 @@
+# frozen_string_literal: true
+
+class Settings::BackgroundJobsController < Admin::BaseController
+  CANCELLABLE_BASE_TYPES = [ Sync, Import, ImportSession, FamilyExport ].freeze
+
+  def show
+    @breadcrumbs = [
+      [ t("breadcrumbs.home"), root_path ],
+      [ t("settings.background_jobs.show.page_title"), nil ]
+    ]
+
+    @console = BackgroundJobConsole.new
+    @operations = @console.operations
+  end
+
+  def cancel
+    record = find_record!
+    console = BackgroundJobConsole.new
+
+    # Server-side re-check — the button state in the UI is not trusted. The
+    # with_lock block re-reads the record, so a job finishing between render
+    # and click cannot be clobbered (it already moved the status on). The
+    # stuck-window check repeats inside the lock too: cancellable? evaluated
+    # Sidekiq liveness outside the transaction (re-running Redis calls under
+    # a row lock would be worse), so a worker that grabbed the job in between
+    # shows up here as a freshly-touched updated_at.
+    cancelled = console.cancellable?(record) && record.with_lock do
+      if cancellable_status?(record) && record.updated_at <= BackgroundJobConsole::STUCK_AFTER.ago
+        prior_status = record.status
+        apply_cancel!(record)
+        audit_cancel!(record, prior_status)
+        true
+      else
+        false
+      end
+    end
+
+    if cancelled
+      redirect_to settings_background_jobs_path, notice: t(".cancelled", type: record.class.name)
+    else
+      redirect_to settings_background_jobs_path, alert: t(".not_cancellable")
+    end
+  end
+
+  private
+    # Resolves record_type without reflecting on request input: rather than
+    # turning the param into a constant, look the id up in each cancellable
+    # base table and require the claimed type to match the found record's
+    # class (or its base class). STI subclass names are still accepted —
+    # the UI sends base_class names, but a direct request naming e.g.
+    # TransactionImport shouldn't 404.
+    def find_record!
+      claimed_type = params[:record_type].to_s
+
+      CANCELLABLE_BASE_TYPES.each do |base|
+        record = base.find_by(id: params[:id])
+        return record if record && [ record.class.name, base.name ].include?(claimed_type)
+      end
+
+      raise ActiveRecord::RecordNotFound, "Unknown record type"
+    end
+
+    # User-facing: surfaces as the failed operation's error in the family UI.
+    def cancelled_error_message
+      t("settings.background_jobs.cancel.cancelled_error")
+    end
+
+    def cancellable_status?(record)
+      case record
+      when Sync then record.in_progress?
+      when Import then record.importing? || record.reverting?
+      when ImportSession then record.importing?
+      when FamilyExport then record.pending? || record.processing?
+      end
+    end
+
+    def apply_cancel!(record)
+      case record
+      when Sync
+        record.mark_stale!
+      when PdfImport
+        if record.reverting?
+          # A stuck revert may have half-deleted entries — pending would
+          # present the import as publishable again. Route it through the
+          # same revert_failed retry path as every other import.
+          record.update!(status: :revert_failed, error: cancelled_error_message)
+        else
+          # importing is the AI-processing claim — release it so the user
+          # can re-trigger, mirroring ProcessPdfJob's own reclaim.
+          record.update!(status: :pending)
+        end
+      when Import
+        record.update!(
+          status: record.reverting? ? :revert_failed : :failed,
+          error: cancelled_error_message
+        )
+      when ImportSession
+        record.update!(
+          status: :failed,
+          error_details: { "code" => "cancelled_by_admin", "message" => cancelled_error_message }
+        )
+      when FamilyExport
+        record.update!(status: :failed)
+      end
+    end
+
+    def audit_cancel!(record, prior_status)
+      DebugLogEntry.capture(
+        category: "background_jobs",
+        level: "warn",
+        message: "#{record.class.name} #{record.id} marked as lost from the background jobs console (was #{prior_status})",
+        source: self.class.name,
+        family: BackgroundJobConsole.family_for(record),
+        metadata: {
+          record_type: record.class.name,
+          record_id: record.id,
+          previous_status: prior_status,
+          new_status: record.status,
+          actor_user_id: Current.user.id
+        }
+      )
+    end
+end

app/models/background_job_console.rb

@@ -0,0 +1,180 @@
+require "sidekiq/api"
+
+# Backs the super-admin background jobs console (/settings/background_jobs).
+#
+# Combines domain truth (in-flight Sync / Import / ImportSession / FamilyExport
+# records across all families) with Sidekiq runtime truth (worker processes,
+# queue depths, jobs currently executing) so an operator can tell a running
+# job from one whose worker died and mark the latter as lost.
+#
+# Unlike BackgroundJobHealth this fails CLOSED: when Redis is unreachable,
+# liveness is unknown and destructive actions are refused rather than the
+# console pretending everything is healthy.
+class BackgroundJobConsole
+  OPERATIONS_LIMIT = 100
+
+  # A record younger than this may belong to a job that is merely queued
+  # behind a backlog or between Sidekiq heartbeats — refuse to touch it.
+  STUCK_AFTER = 30.minutes
+
+  # Upper bound on queue/retry/schedule entries scanned for record
+  # references. Past this the backlog is inspected only partially, so
+  # liveness is unknowable and cancellation fails closed (like redis_error?).
+  QUEUE_SCAN_LIMIT = 5_000
+
+  Stats = Struct.new(:processes, :busy, :enqueued, :retry_size, :dead_size, :scheduled_size, :queues, keyword_init: true)
+
+  attr_reader :stats
+
+  def initialize
+    @redis_error = false
+    @queue_scan_truncated = false
+    @running_global_ids = Set.new
+    @queued_global_ids = Set.new
+    @stats = nil
+    load_runtime_state
+  end
+
+  def redis_error?
+    @redis_error
+  end
+
+  def queue_scan_truncated?
+    @queue_scan_truncated
+  end
+
+  # In-flight operations across ALL families, newest activity first. This is
+  # deliberately instance-global — the console is super-admin only.
+  def operations
+    @operations ||= [
+      Sync.incomplete.includes(:syncable).order(updated_at: :desc).limit(OPERATIONS_LIMIT).to_a,
+      Import.where(status: [ :importing, :reverting ]).includes(:family).order(updated_at: :desc).limit(OPERATIONS_LIMIT).to_a,
+      ImportSession.where(status: :importing).includes(:family).order(updated_at: :desc).limit(OPERATIONS_LIMIT).to_a,
+      FamilyExport.where(status: [ :pending, :processing ]).includes(:family).order(updated_at: :desc).limit(OPERATIONS_LIMIT).to_a
+    ].flatten.sort_by(&:updated_at).reverse
+  end
+
+  # A record's job is visibly executing right now (its GlobalID appears in a
+  # worker's payload). False also means "unknown" when redis_error? is set —
+  # callers must check that first for destructive decisions.
+  def running?(record)
+    @running_global_ids.include?(record.to_global_id.to_s)
+  end
+
+  # A job referencing this record is sitting in a queue, the retry set, or
+  # the scheduled set. Such a job WILL run later — most of the affected job
+  # classes don't abort just because an operator flipped the record's status,
+  # so terminalizing now would invite duplicate/conflicting work when it
+  # finally executes.
+  def enqueued?(record)
+    @queued_global_ids.include?(record.to_global_id.to_s)
+  end
+
+  # Safe to force-terminalize: liveness is knowable (Redis reachable, backlog
+  # scan complete), no job referencing the record is executing or waiting to
+  # execute, the record has been idle past the stuck window, and (for syncs)
+  # no children are still in flight — a parent Sync legitimately has no live
+  # job of its own while its children run.
+  def cancellable?(record)
+    return false if redis_error?
+    return false if queue_scan_truncated?
+    return false if running?(record)
+    return false if enqueued?(record)
+    return false if record.updated_at > STUCK_AFTER.ago
+    return false if record.is_a?(Sync) && record.children.incomplete.exists?
+
+    true
+  end
+
+  def self.family_for(record)
+    if record.is_a?(Sync)
+      syncable = record.syncable
+      syncable.is_a?(Family) ? syncable : syncable&.family
+    else
+      record.family
+    end
+  end
+
+  private
+    def load_runtime_state
+      processes = Sidekiq::ProcessSet.new
+      sidekiq_stats = Sidekiq::Stats.new
+
+      @stats = Stats.new(
+        processes: processes.size,
+        busy: processes.sum { |process| process["busy"].to_i },
+        enqueued: sidekiq_stats.enqueued,
+        retry_size: sidekiq_stats.retry_size,
+        dead_size: sidekiq_stats.dead_size,
+        scheduled_size: sidekiq_stats.scheduled_size,
+        queues: Sidekiq::Queue.all.map { |queue| { name: queue.name, size: queue.size, latency: queue.latency.round(1) } }
+      )
+
+      @running_global_ids = collect_running_global_ids
+      @queued_global_ids = collect_queued_global_ids
+    rescue => e
+      Rails.logger.warn("BackgroundJobConsole: Sidekiq state unavailable: #{e.class}: #{e.message}")
+      @redis_error = true
+      @stats = nil
+      @running_global_ids = Set.new
+      @queued_global_ids = Set.new
+    end
+
+    # All jobs are ActiveJob-wrapped, so record references appear in worker
+    # payloads as serialized GlobalIDs ({"_aj_globalid" => "gid://..."}).
+    def collect_running_global_ids
+      ids = Set.new
+
+      Sidekiq::Workers.new.each do |_process_id, _thread_id, work|
+        payload = work.respond_to?(:payload) ? work.payload : work["payload"]
+        payload = JSON.parse(payload) if payload.is_a?(String)
+        collect_global_ids(payload, ids)
+      rescue JSON::ParserError
+        next
+      end
+
+      ids
+    end
+
+    # Record references in jobs that are waiting to run: queue backlogs, the
+    # retry set, and the scheduled set. Bounded by QUEUE_SCAN_LIMIT — on a
+    # truncated scan, cancellation fails closed via queue_scan_truncated?.
+    def collect_queued_global_ids
+      ids = Set.new
+      scanned = 0
+
+      each_waiting_job do |item|
+        scanned += 1
+        if scanned > QUEUE_SCAN_LIMIT
+          @queue_scan_truncated = true
+          break
+        end
+        collect_global_ids(item, ids)
+      end
+
+      ids
+    end
+
+    def each_waiting_job(&block)
+      Sidekiq::Queue.all.each do |queue|
+        queue.each { |job| yield job.item }
+      end
+      Sidekiq::RetrySet.new.each { |job| yield job.item }
+      Sidekiq::ScheduledSet.new.each { |job| yield job.item }
+    end
+
+    def collect_global_ids(node, ids)
+      case node
+      when Hash
+        node.each do |key, value|
+          if key == "_aj_globalid" && value.is_a?(String)
+            ids << value
+          else
+            collect_global_ids(value, ids)
+          end
+        end
+      when Array
+        node.each { |value| collect_global_ids(value, ids) }
+      end
+    end
+end

app/views/settings/_settings_nav.html.erb

@@ -32,6 +32,7 @@ nav_sections = [
         { label: t(".api_keys_label"), path: settings_api_keys_path, icon: "key" },
         { label: t(".mcp_label"), path: settings_mcp_path, icon: "plug" },
         { label: t(".debug_label", default: "Debug"), path: settings_debug_path, icon: "bug", if: Current.user&.super_admin? },
+        { label: t(".background_jobs_label"), path: settings_background_jobs_path, icon: "list-checks", if: Current.user&.super_admin? },
         { label: t(".self_hosting_label"), path: settings_hosting_path, icon: "database", if: self_hosted? },
         { label: t(".imports_label"), path: imports_path, icon: "download" },
         { label: t(".exports_label"), path: family_exports_path, icon: "upload" },

app/views/settings/background_jobs/_operation.html.erb

@@ -0,0 +1,49 @@
+<% family = BackgroundJobConsole.family_for(operation) %>
+<% running = console.running?(operation) %>
+<tr>
+  <td class="px-4 py-3 text-sm text-primary whitespace-nowrap">
+    <%= operation.is_a?(Sync) ? t(".sync_label", type: operation.syncable_type) : operation.class.name %>
+    <p class="text-xs text-secondary font-mono" title="<%= operation.id %>"><%= operation.id.to_s.split("-").first %></p>
+  </td>
+  <td class="px-4 py-3 text-sm text-primary whitespace-nowrap"><%= family&.name || t(".missing_value") %></td>
+  <td class="px-4 py-3 text-sm text-primary whitespace-nowrap"><%= operation.status %></td>
+  <td class="px-4 py-3 text-sm text-secondary whitespace-nowrap">
+    <%= t(".ago", time: time_ago_in_words(operation.updated_at)) %>
+  </td>
+  <td class="px-4 py-3 text-sm whitespace-nowrap">
+    <% if running %>
+      <span class="flex items-center gap-2 text-primary">
+        <span class="animate-spin h-3 w-3 border-2 border-secondary border-t-transparent rounded-full"></span>
+        <%= t(".running") %>
+      </span>
+    <% elsif console.redis_error? %>
+      <span class="text-secondary"><%= t(".unknown") %></span>
+    <% elsif console.enqueued?(operation) %>
+      <span class="text-secondary"><%= t(".queued") %></span>
+    <% elsif operation.updated_at > BackgroundJobConsole::STUCK_AFTER.ago %>
+      <span class="text-secondary"><%= t(".recent") %></span>
+    <% else %>
+      <span class="flex items-center gap-2 text-primary">
+        <%= icon "alert-triangle", class: "w-4 h-4 text-warning" %>
+        <%= t(".stuck") %>
+      </span>
+    <% end %>
+  </td>
+  <td class="px-4 py-3 text-sm whitespace-nowrap text-right">
+    <% if console.cancellable?(operation) %>
+      <% action_key = case operation
+         when Sync then "mark_stale"
+         when PdfImport then "release_claim"
+         else "mark_failed"
+         end %>
+      <%= button_to cancel_settings_background_jobs_path(record_type: operation.class.base_class.name, id: operation.id),
+                    method: :post,
+                    class: "text-sm text-destructive hover:underline",
+                    data: { turbo_confirm: t(".confirm"), turbo_frame: "_top" } do %>
+        <%= t(".actions.#{action_key}") %>
+      <% end %>
+    <% else %>
+      <span class="text-xs text-secondary"><%= t(".missing_value") %></span>
+    <% end %>
+  </td>
+</tr>

app/views/settings/background_jobs/show.html.erb

@@ -0,0 +1,68 @@
+<%= content_for :page_title, t(".page_title") %>
+
+<%= turbo_frame_tag "background_jobs_console",
+    data: {
+      controller: "polling",
+      polling_url_value: settings_background_jobs_path,
+      polling_interval_value: 10000
+    } do %>
+  <div class="space-y-4">
+    <%= settings_section title: t(".title"), subtitle: t(".subtitle") do %>
+      <% if @console.redis_error? || @console.stats.nil? %>
+        <%= render DS::Alert.new(variant: :warning, message: t(".redis_unreachable")) %>
+      <% else %>
+        <div class="space-y-3">
+          <div class="grid grid-cols-2 sm:grid-cols-3 lg:grid-cols-6 gap-3">
+            <% {
+              workers: @console.stats.processes,
+              busy: @console.stats.busy,
+              enqueued: @console.stats.enqueued,
+              retries: @console.stats.retry_size,
+              scheduled: @console.stats.scheduled_size,
+              dead: @console.stats.dead_size
+            }.each do |label, value| %>
+              <div class="rounded-lg bg-container-inset p-3">
+                <p class="text-xs text-secondary uppercase"><%= t(".stats.#{label}") %></p>
+                <p class="text-lg font-medium text-primary"><%= value %></p>
+              </div>
+            <% end %>
+          </div>
+
+          <div class="flex flex-wrap gap-2">
+            <% @console.stats.queues.each do |queue| %>
+              <span class="rounded-full bg-container-inset px-3 py-1 text-xs text-secondary font-mono">
+                <%= queue[:name] %>: <%= queue[:size] %> · <%= t(".stats.latency", value: queue[:latency]) %>
+              </span>
+            <% end %>
+          </div>
+        </div>
+      <% end %>
+    <% end %>
+
+    <%= settings_section title: t(".operations_title"), subtitle: t(".operations_subtitle") do %>
+      <% if @operations.any? %>
+        <div class="overflow-x-auto">
+          <table class="w-full">
+            <thead class="bg-surface border-b border-primary">
+              <tr>
+                <th class="px-4 py-3 text-left text-xs font-medium text-secondary uppercase"><%= t(".table.type") %></th>
+                <th class="px-4 py-3 text-left text-xs font-medium text-secondary uppercase"><%= t(".table.family") %></th>
+                <th class="px-4 py-3 text-left text-xs font-medium text-secondary uppercase"><%= t(".table.status") %></th>
+                <th class="px-4 py-3 text-left text-xs font-medium text-secondary uppercase"><%= t(".table.last_activity") %></th>
+                <th class="px-4 py-3 text-left text-xs font-medium text-secondary uppercase"><%= t(".table.liveness") %></th>
+                <th class="px-4 py-3 text-right text-xs font-medium text-secondary uppercase"><%= t(".table.action") %></th>
+              </tr>
+            </thead>
+            <tbody class="divide-y divide-alpha-black-200 theme-dark:divide-alpha-white-200">
+              <% @operations.each do |operation| %>
+                <%= render "settings/background_jobs/operation", operation: operation, console: @console %>
+              <% end %>
+            </tbody>
+          </table>
+        </div>
+      <% else %>
+        <div class="p-8 text-center text-secondary"><%= t(".empty") %></div>
+      <% end %>
+    <% end %>
+  </div>
+<% end %>

config/initializers/rack_attack.rb

@@ -28,6 +28,13 @@ class Rack::Attack
     request.ip if request.path.start_with?("/admin/")
   end
 
+  # The background jobs console lives under /settings (so its polling GET
+  # isn't throttled), but its mutation is destructive and super-admin only —
+  # rate limit it independently.
+  throttle("background_jobs_console/ip", limit: 30, period: 1.minute) do |request|
+    request.ip if request.post? && request.path == "/settings/background_jobs/cancel"
+  end
+
   # Determine limits based on self-hosted mode
   self_hosted = Rails.application.config.app_mode.self_hosted?
 

config/locales/views/settings/en.yml

@@ -6,6 +6,49 @@ en:
         renewal: "Your contribution continues on %{date}."
         cancellation: "Your contribution ends on %{date}."
   settings:
+    background_jobs:
+      show:
+        page_title: "Background jobs"
+        title: "Worker status"
+        subtitle: "Live Sidekiq runtime state: worker processes, queue depths, and retry/dead sets."
+        redis_unreachable: "Sidekiq state is unavailable (Redis unreachable). Liveness is unknown, so actions are disabled."
+        operations_title: "In-flight operations"
+        operations_subtitle: "Syncs, imports, and exports across all families that have not reached a terminal status. Operations idle for over 30 minutes with no visible job are presumed lost and can be marked as such."
+        empty: "No in-flight operations."
+        missing_value: "-"
+        stats:
+          workers: "Workers"
+          busy: "Busy"
+          enqueued: "Enqueued"
+          retries: "Retries"
+          scheduled: "Scheduled"
+          dead: "Dead"
+          latency: "%{value}s latency"
+        table:
+          type: "Operation"
+          family: "Family"
+          status: "Status"
+          last_activity: "Last activity"
+          liveness: "Liveness"
+          action: "Action"
+      operation:
+        ago: "%{time} ago"
+        sync_label: "Sync · %{type}"
+        running: "Running"
+        queued: "Queued"
+        recent: "Recent"
+        unknown: "Unknown"
+        stuck: "Stuck — presumed lost"
+        confirm: "Mark this operation as lost? The record is moved to a terminal status so it can be retried. The underlying data is not touched."
+        missing_value: "-"
+        actions:
+          mark_stale: "Mark stale"
+          mark_failed: "Mark failed"
+          release_claim: "Release claim"
+      cancel:
+        cancelled: "%{type} marked as lost."
+        not_cancellable: "This operation can't be modified right now — its job may still be queued or running. Try again once it has been idle for 30 minutes."
+        cancelled_error: "Marked as failed by an administrator — the background job was presumed lost."
     debugs:
       show:
         page_title: "Debug"
@@ -266,6 +309,7 @@ en:
       accounts_label: Accounts
       advanced_section_title: Advanced
       ai_prompts_label: AI Prompts
+      background_jobs_label: Background jobs
       api_key_label: API Keys
       payment_label: Payment
       categories_label: Categories

config/routes.rb

@@ -299,6 +299,9 @@ Rails.application.routes.draw do
     resource :preferences, only: %i[show update]
     resource :appearance, only: %i[show update]
     resource :debug, only: :show
+    resource :background_jobs, controller: "background_jobs", only: :show do
+      post :cancel
+    end
     resource :hosting, only: %i[show update] do
       delete :clear_cache, on: :collection
       delete :disconnect_external_assistant, on: :collection

test/controllers/settings/background_jobs_controller_test.rb

@@ -0,0 +1,205 @@
+require "test_helper"
+
+class Settings::BackgroundJobsControllerTest < ActionDispatch::IntegrationTest
+  setup do
+    stub_sidekiq
+  end
+
+  test "super admin can view the console" do
+    sign_in users(:sure_support_staff)
+
+    get settings_background_jobs_path
+
+    assert_response :success
+  end
+
+  test "console renders in-flight operations with actions" do
+    sign_in users(:sure_support_staff)
+
+    stuck = imports(:transaction)
+    stuck.update_columns(status: "importing", updated_at: 1.hour.ago)
+
+    fresh_sync = Sync.create!(syncable: accounts(:depository), status: :syncing)
+
+    get settings_background_jobs_path
+
+    assert_response :success
+    assert_match stuck.id, response.body
+    assert_match fresh_sync.id, response.body
+    assert_match I18n.t("settings.background_jobs.operation.actions.mark_failed"), response.body
+  end
+
+  test "family admin is redirected away" do
+    sign_in users(:family_admin)
+
+    get settings_background_jobs_path
+
+    assert_redirected_to root_path
+  end
+
+  test "member is redirected away" do
+    sign_in users(:family_member)
+
+    get settings_background_jobs_path
+
+    assert_redirected_to root_path
+  end
+
+  test "cancel marks a stuck import as failed and writes an audit entry" do
+    sign_in users(:sure_support_staff)
+
+    import = imports(:transaction)
+    import.update_columns(status: "importing", updated_at: 1.hour.ago)
+
+    assert_difference "DebugLogEntry.count", 1 do
+      post cancel_settings_background_jobs_path(record_type: "Import", id: import.id)
+    end
+
+    assert_redirected_to settings_background_jobs_path
+    assert_equal "failed", import.reload.status
+
+    entry = DebugLogEntry.order(:created_at).last
+    assert_equal "background_jobs", entry.category
+    assert_equal users(:sure_support_staff).id, entry.metadata["actor_user_id"]
+  end
+
+  test "cancel releases a stuck PdfImport claim back to pending" do
+    sign_in users(:sure_support_staff)
+
+    pdf = imports(:pdf)
+    pdf.update_columns(status: "importing", updated_at: 1.hour.ago)
+
+    post cancel_settings_background_jobs_path(record_type: "Import", id: pdf.id)
+
+    assert_equal "pending", pdf.reload.status
+  end
+
+  test "cancel marks a stuck sync as stale" do
+    sign_in users(:sure_support_staff)
+
+    sync = Sync.create!(syncable: accounts(:depository), status: :syncing)
+    sync.update_columns(updated_at: 1.hour.ago)
+
+    post cancel_settings_background_jobs_path(record_type: "Sync", id: sync.id)
+
+    assert_equal "stale", sync.reload.status
+  end
+
+  test "cancel refuses a record inside the stuck window" do
+    sign_in users(:sure_support_staff)
+
+    import = imports(:transaction)
+    import.update_columns(status: "importing", updated_at: 1.minute.ago)
+
+    post cancel_settings_background_jobs_path(record_type: "Import", id: import.id)
+
+    assert_equal "importing", import.reload.status
+    assert_equal I18n.t("settings.background_jobs.cancel.not_cancellable"), flash[:alert]
+  end
+
+  test "cancel refuses when the record's job is visibly running" do
+    sign_in users(:sure_support_staff)
+
+    import = imports(:transaction)
+    import.update_columns(status: "importing", updated_at: 1.hour.ago)
+
+    stub_sidekiq(worker_payloads: [
+      { "wrapped" => "ImportJob", "args" => [ { "arguments" => [ { "_aj_globalid" => import.to_global_id.to_s } ] } ] }
+    ])
+
+    post cancel_settings_background_jobs_path(record_type: "Import", id: import.id)
+
+    assert_equal "importing", import.reload.status
+  end
+
+  test "cancel resolves STI subclass record types" do
+    sign_in users(:sure_support_staff)
+
+    import = imports(:transaction)
+    import.update_columns(status: "importing", updated_at: 1.hour.ago)
+
+    post cancel_settings_background_jobs_path(record_type: "TransactionImport", id: import.id)
+
+    assert_redirected_to settings_background_jobs_path
+    assert_equal "failed", import.reload.status
+  end
+
+  test "cancel routes a stuck PdfImport revert to revert_failed, not pending" do
+    sign_in users(:sure_support_staff)
+
+    pdf = imports(:pdf)
+    pdf.update_columns(status: "reverting", updated_at: 1.hour.ago)
+
+    post cancel_settings_background_jobs_path(record_type: "Import", id: pdf.id)
+
+    assert_equal "revert_failed", pdf.reload.status
+  end
+
+  test "cancel refuses when the record's job is waiting in a queue" do
+    sign_in users(:sure_support_staff)
+
+    import = imports(:transaction)
+    import.update_columns(status: "importing", updated_at: 1.hour.ago)
+
+    queue = mock("Queue")
+    queue.stubs(name: "default", size: 1, latency: 0.0)
+    queue.stubs(:each).multiple_yields([ stub(item: { "wrapped" => "ImportJob", "args" => [ { "arguments" => [ { "_aj_globalid" => import.to_global_id.to_s } ] } ] }) ])
+    Sidekiq::Queue.stubs(:all).returns([ queue ])
+
+    post cancel_settings_background_jobs_path(record_type: "Import", id: import.id)
+
+    assert_equal "importing", import.reload.status
+    assert_equal I18n.t("settings.background_jobs.cancel.not_cancellable"), flash[:alert]
+  end
+
+  test "cancel refuses unknown record types" do
+    sign_in users(:sure_support_staff)
+
+    post cancel_settings_background_jobs_path(record_type: "User", id: users(:family_admin).id)
+
+    assert_response :not_found
+  end
+
+  test "cancel requires super admin" do
+    sign_in users(:family_admin)
+
+    import = imports(:transaction)
+    import.update_columns(status: "importing", updated_at: 1.hour.ago)
+
+    post cancel_settings_background_jobs_path(record_type: "Import", id: import.id)
+
+    assert_redirected_to root_path
+    assert_equal "importing", import.reload.status
+  end
+
+  private
+    def stub_sidekiq(worker_payloads: [])
+      process_set = mock("ProcessSet")
+      process_set.stubs(:size).returns(1)
+      process_set.stubs(:sum).returns(worker_payloads.size)
+      Sidekiq::ProcessSet.stubs(:new).returns(process_set)
+
+      stats = mock("Stats")
+      stats.stubs(:enqueued).returns(0)
+      stats.stubs(:retry_size).returns(0)
+      stats.stubs(:dead_size).returns(0)
+      stats.stubs(:scheduled_size).returns(0)
+      Sidekiq::Stats.stubs(:new).returns(stats)
+
+      Sidekiq::Queue.stubs(:all).returns([])
+
+      empty_set = mock("JobSet")
+      empty_set.stubs(:each)
+      Sidekiq::RetrySet.stubs(:new).returns(empty_set)
+      Sidekiq::ScheduledSet.stubs(:new).returns(empty_set)
+
+      workers = mock("Workers")
+      yields = worker_payloads.map { |payload| [ "process", "thread", { "payload" => payload.to_json } ] }
+      if yields.any?
+        workers.stubs(:each).multiple_yields(*yields)
+      else
+        workers.stubs(:each)
+      end
+      Sidekiq::Workers.stubs(:new).returns(workers)
+    end
+end

test/models/background_job_console_test.rb

@@ -0,0 +1,121 @@
+require "test_helper"
+
+class BackgroundJobConsoleTest < ActiveSupport::TestCase
+  test "fails closed when Sidekiq/Redis is unreachable" do
+    Sidekiq::ProcessSet.stubs(:new).raises(StandardError.new("no redis"))
+
+    console = BackgroundJobConsole.new
+    sync = Sync.create!(syncable: accounts(:depository), status: :syncing)
+    sync.update_columns(updated_at: 1.hour.ago)
+
+    assert console.redis_error?
+    assert_nil console.stats
+    assert_not console.cancellable?(sync)
+  end
+
+  test "detects running records via GlobalIDs in worker payloads" do
+    sync = Sync.create!(syncable: accounts(:depository), status: :syncing)
+    sync.update_columns(updated_at: 1.hour.ago)
+
+    stub_sidekiq(worker_payloads: [
+      { "wrapped" => "SyncJob", "args" => [ { "arguments" => [ { "_aj_globalid" => sync.to_global_id.to_s } ] } ] }
+    ])
+
+    console = BackgroundJobConsole.new
+
+    assert console.running?(sync)
+    assert_not console.cancellable?(sync)
+  end
+
+  test "cancellable only when idle past the stuck window with no live job" do
+    stub_sidekiq
+
+    console = BackgroundJobConsole.new
+
+    fresh = Sync.create!(syncable: accounts(:depository), status: :syncing)
+    stuck = Sync.create!(syncable: accounts(:connected), status: :syncing)
+    stuck.update_columns(updated_at: 1.hour.ago)
+
+    assert_not console.cancellable?(fresh)
+    assert console.cancellable?(stuck)
+  end
+
+  test "a record referenced by a queued job is not cancellable" do
+    sync = Sync.create!(syncable: accounts(:depository), status: :syncing)
+    sync.update_columns(updated_at: 1.hour.ago)
+
+    stub_sidekiq(queued_items: [
+      { "wrapped" => "SyncJob", "args" => [ { "arguments" => [ { "_aj_globalid" => sync.to_global_id.to_s } ] } ] }
+    ])
+
+    console = BackgroundJobConsole.new
+
+    assert console.enqueued?(sync)
+    assert_not console.running?(sync)
+    assert_not console.cancellable?(sync)
+  end
+
+  test "a truncated backlog scan fails closed" do
+    sync = Sync.create!(syncable: accounts(:depository), status: :syncing)
+    sync.update_columns(updated_at: 1.hour.ago)
+
+    filler = Array.new(BackgroundJobConsole::QUEUE_SCAN_LIMIT + 1) { { "args" => [] } }
+    stub_sidekiq(queued_items: filler)
+
+    console = BackgroundJobConsole.new
+
+    assert console.queue_scan_truncated?
+    assert_not console.cancellable?(sync)
+  end
+
+  test "a parent sync with incomplete children is not cancellable" do
+    stub_sidekiq
+
+    parent = Sync.create!(syncable: families(:dylan_family), status: :syncing)
+    Sync.create!(syncable: accounts(:depository), status: :syncing, parent: parent)
+    parent.update_columns(updated_at: 1.hour.ago)
+
+    console = BackgroundJobConsole.new
+
+    assert_not console.cancellable?(parent)
+  end
+
+  private
+    def stub_sidekiq(worker_payloads: [], queued_items: [])
+      process_set = mock("ProcessSet")
+      process_set.stubs(:size).returns(1)
+      process_set.stubs(:sum).returns(worker_payloads.size)
+      Sidekiq::ProcessSet.stubs(:new).returns(process_set)
+
+      stats = mock("Stats")
+      stats.stubs(:enqueued).returns(queued_items.size)
+      stats.stubs(:retry_size).returns(0)
+      stats.stubs(:dead_size).returns(0)
+      stats.stubs(:scheduled_size).returns(0)
+      Sidekiq::Stats.stubs(:new).returns(stats)
+
+      if queued_items.any?
+        queue = mock("Queue")
+        queue.stubs(name: "default", size: queued_items.size, latency: 0.0)
+        queue_yields = queued_items.map { |item| [ stub(item: item) ] }
+        queue.stubs(:each).multiple_yields(*queue_yields)
+        Sidekiq::Queue.stubs(:all).returns([ queue ])
+      else
+        Sidekiq::Queue.stubs(:all).returns([])
+      end
+
+      empty_set = mock("JobSet")
+      empty_set.stubs(:each)
+      Sidekiq::RetrySet.stubs(:new).returns(empty_set)
+      Sidekiq::ScheduledSet.stubs(:new).returns(empty_set)
+
+      workers = mock("Workers")
+      yields = worker_payloads.map { |payload| [ "process", "thread", { "payload" => payload.to_json } ] }
+      if yields.any?
+        workers.stubs(:each).multiple_yields(*yields)
+      else
+        workers.stubs(:each)
+      end
+      Sidekiq::Workers.stubs(:new).returns(workers)
+    end
+end