diff --git a/app/jobs/application_job.rb b/app/jobs/application_job.rb
new file mode 100644
index 00000000000..d394c3d1062
--- /dev/null
+++ b/app/jobs/application_job.rb
@@ -0,0 +1,7 @@
+class ApplicationJob < ActiveJob::Base
+  # Automatically retry jobs that encountered a deadlock
+  # retry_on ActiveRecord::Deadlocked
+
+  # Most jobs are safe to ignore if the underlying records are no longer available
+  # discard_on ActiveJob::DeserializationError
+end
diff --git a/app/jobs/vendor_proof_job.rb b/app/jobs/vendor_proof_job.rb
new file mode 100644
index 00000000000..7977fe142dd
--- /dev/null
+++ b/app/jobs/vendor_proof_job.rb
@@ -0,0 +1,12 @@
+class VendorProofJob < ApplicationJob
+  queue_as :default
+
+  def perform(document_capture_session_id, stages)
+    # binding.pry
+    dcs = DocumentCaptureSession.find_by(uuid: document_capture_session_id)
+    result = dcs.load_proofing_result
+    stages = stages.map(&:to_sym)
+    idv_result = Idv::Agent.new(result.pii).proof(*stages)
+    dcs.store_proofing_result(result.pii, idv_result)
+  end
+end
diff --git a/app/models/document_capture_session.rb b/app/models/document_capture_session.rb
index ee3e26d5cd3..a27f25ae1a7 100644
--- a/app/models/document_capture_session.rb
+++ b/app/models/document_capture_session.rb
@@ -7,6 +7,10 @@ def load_result
     DocumentCaptureSessionResult.load(result_id)
   end
 
+  def load_proofing_result
+    ProofingDocumentCaptureSessionResult.load(result_id)
+  end
+
   def store_result_from_response(doc_auth_response)
     DocumentCaptureSessionResult.store(
       id: generate_result_id,
@@ -16,6 +20,23 @@ def store_result_from_response(doc_auth_response)
     save!
   end
 
+  def store_proofing_pii_from_doc(pii_from_doc)
+    ProofingDocumentCaptureSessionResult.store(
+      id: generate_result_id,
+      pii: pii_from_doc,
+      result: nil,
+    )
+    save!
+  end
+
+  def store_proofing_result(pii_from_doc, result)
+    ProofingDocumentCaptureSessionResult.store(
+      id: result_id,
+      pii: pii_from_doc,
+      result: result,
+    )
+  end
+
   def expired?
     return true unless requested_at
     requested_at + Figaro.env.doc_capture_request_valid_for_minutes.to_i.minutes < Time.zone.now
diff --git a/app/services/flow/base_flow.rb b/app/services/flow/base_flow.rb
index 6f403151336..43fb3eac549 100644
--- a/app/services/flow/base_flow.rb
+++ b/app/services/flow/base_flow.rb
@@ -26,15 +26,18 @@ def redirect_to(url)
     def handle(step)
       @flow_session[:error_message] = nil
       @flow_session[:notice] = nil
-      handler = steps[step] || actions[step]
-      return failure("Unhandled step #{step}") unless handler
-      wrap_send(handler)
+      return failure("Unhandled step #{step}") unless handler(step)
+      wrap_send(step)
+    end
+
+    def handler(step)
+      steps[step] || actions[step]
     end
 
     private
 
-    def wrap_send(handler)
-      obj = handler.new(self)
+    def wrap_send(step)
+      obj = handler(step).new(self)
       value = obj.base_call
       form_response(obj, value)
     end
diff --git a/app/services/flow/base_step.rb b/app/services/flow/base_step.rb
index dfd5b3db31e..e7cf04a9522 100644
--- a/app/services/flow/base_step.rb
+++ b/app/services/flow/base_step.rb
@@ -26,6 +26,10 @@ def mark_step_incomplete(step = nil)
       flow_session.delete(klass.to_s)
     end
 
+    def async?
+      false
+    end
+
     def self.acceptable_response_object?(obj)
       obj.is_a?(FormResponse) || obj.is_a?(DocAuth::Response)
     end
diff --git a/app/services/flow/flow_state_machine.rb b/app/services/flow/flow_state_machine.rb
index 735fa09d55e..70c41b1e833 100644
--- a/app/services/flow/flow_state_machine.rb
+++ b/app/services/flow/flow_state_machine.rb
@@ -14,24 +14,82 @@ def index
     end
 
     def show
-      step = current_step
-      analytics.track_event(analytics_visited, step: step) if @analytics_id
-      Funnel::DocAuth::RegisterStep.new(user_id, issuer).call(step, :view, true)
-      register_campaign
-      render_step(step, flow.flow_session)
+      flow_handler = flow.handler(current_step).new(flow)
+
+      if flow_handler.async?
+        # binding.pry
+        async_show(flow_handler)
+      else
+        begin_step
+      end
     end
 
     def update
-      step = current_step
-      result = flow.handle(step)
-      analytics.track_event(analytics_submitted, result.to_h.merge(step: step)) if @analytics_id
-      register_update_step(step, result)
-      flow_finish and return unless next_step
-      render_update(step, result)
+      flow_handler = flow.handler(current_step).new(flow)
+
+      if flow_handler.async?
+        async_update(flow_handler)
+      else
+        result = flow.handle(current_step)
+        end_step(result)
+      end
     end
 
     private
 
+    def async_show(flow_handler)
+      async_state = flow_handler.async_state
+      case async_state.status
+      when :none
+        begin_step
+      when :in_progress
+        redirect_to send(@step_url, step: current_step)
+      when :timed_out
+        begin_step
+      when :done
+        result = flow_handler.after_call(async_state.pii, async_state.result)
+        flow_handler.mark_step_complete(current_step) if result.success?
+        flow_handler.delete_async unless result.success?
+        end_step(result)
+      end
+    end
+
+    def async_update(flow_handler)
+      case flow_handler.async_state.status
+      when :none
+        flow.handle(current_step)
+        # binding.pry
+        flow_handler.mark_step_incomplete(current_step)
+        redirect_to send(@step_url, step: current_step)
+      when :in_progress
+        redirect_to send(@step_url, step: current_step)
+      when :timed_out
+        flow.handle(current_step)
+        flow_handler.mark_step_incomplete(current_step)
+        redirect_to send(@step_url, step: current_step)
+      when :done
+        redirect_to send(@step_url, step: current_step)
+      end
+    end
+
+    def begin_step
+      analytics.track_event(analytics_visited, step: current_step) if @analytics_id
+      Funnel::DocAuth::RegisterStep.new(user_id, issuer).call(current_step, :view, true)
+      register_campaign
+      render_step(current_step, flow.flow_session)
+    end
+
+    def end_step(result)
+      # binding.pry
+      if @analytics_id
+        analytics.track_event(analytics_submitted, result.to_h.merge(step: current_step))
+      end
+      register_update_step(current_step, result)
+      flow_finish and return unless next_step
+
+      render_update(current_step, result)
+    end
+
     def current_step
       params[:step]&.underscore
     end
@@ -58,10 +116,15 @@ def issuer
 
     def fsm_initialize
       klass = self.class
+      flow = klass::FSM_SETTINGS[:flow]
+      @step_url = klass::FSM_SETTINGS[:step_url]
+      @final_url = klass::FSM_SETTINGS[:final_url]
+      @analytics_id = klass::FSM_SETTINGS[:analytics_id]
+      @view = klass::FSM_SETTINGS[:view]
       @name = klass.name.underscore.gsub('_controller', '')
-      klass::FSM_SETTINGS.each { |key, value| instance_variable_set("@#{key}", value) }
+
       current_session[@name] ||= {}
-      @flow = @flow.new(self, current_session, @name)
+      @flow = flow.new(self, current_session, @name)
     end
 
     def render_update(step, result)
diff --git a/app/services/idv/session.rb b/app/services/idv/session.rb
index 5dc1dafd217..e3ecee3bd00 100644
--- a/app/services/idv/session.rb
+++ b/app/services/idv/session.rb
@@ -1,8 +1,6 @@
 module Idv
   class Session
     VALID_SESSION_ATTRIBUTES = %i[
-      async_result_id
-      async_result_started_at
       address_verification_mechanism
       applicant
       vendor_phone_confirmation
@@ -15,7 +13,6 @@ class Session
       profile_step_params
       personal_key
       resolution_successful
-      selected_jurisdiction
       step_attempts
     ].freeze
 
diff --git a/app/services/idv/steps/verify_base_step.rb b/app/services/idv/steps/verify_base_step.rb
index 0896e7a54b0..36f38e53b7a 100644
--- a/app/services/idv/steps/verify_base_step.rb
+++ b/app/services/idv/steps/verify_base_step.rb
@@ -11,9 +11,11 @@ class VerifyBaseStep < DocAuthBaseStep
       def perform_resolution_and_check_ssn
         pii_from_doc = flow_session[:pii_from_doc]
         # do resolution first to prevent ssn time/discovery. resolution time order > than db call
-        result = perform_resolution(pii_from_doc)
-        result = check_ssn(pii_from_doc) if result.success?
-        summarize_result_and_throttle_failures(result)
+        idv_result = perform_resolution(pii_from_doc)
+        add_proofing_costs(idv_result)
+        response = idv_result_to_form_response(idv_result)
+        response = check_ssn(pii_from_doc) if response.success?
+        summarize_result_and_throttle_failures(response)
       end
 
       def summarize_result_and_throttle_failures(summary_result)
@@ -53,8 +55,10 @@ def skip_legacy_steps
 
       def perform_resolution(pii_from_doc)
         stages = should_use_aamva?(pii_from_doc) ? %i[resolution state_id] : [:resolution]
-        idv_result = Idv::Agent.new(pii_from_doc).proof(*stages)
-        add_proofing_costs(idv_result)
+        Idv::Agent.new(pii_from_doc).proof(*stages)
+      end
+
+      def idv_result_to_form_response(idv_result)
         FormResponse.new(
           success: idv_success(idv_result),
           errors: idv_errors(idv_result),
@@ -63,6 +67,7 @@ def perform_resolution(pii_from_doc)
       end
 
       def add_proofing_costs(results)
+        # binding.pry
         vendors = results[:context][:stages]
         vendors.each do |hash|
           add_cost(:aamva) if hash[:state_id]
diff --git a/app/services/idv/steps/verify_step.rb b/app/services/idv/steps/verify_step.rb
index a37430041b2..bb155022e0c 100644
--- a/app/services/idv/steps/verify_step.rb
+++ b/app/services/idv/steps/verify_step.rb
@@ -1,8 +1,88 @@
 module Idv
   module Steps
     class VerifyStep < VerifyBaseStep
+      State = Struct.new(:status, :pii, :result, keyword_init: true) do
+        def self.none
+          new(status: :none)
+        end
+
+        def self.timed_out
+          new(status: :timed_out)
+        end
+
+        def self.in_progress
+          new(status: :in_progress)
+        end
+
+        def self.done(pii:, result:)
+          new(status: :done, pii: pii, result: result)
+        end
+
+        private_class_method :new
+      end
+
       def call
-        perform_resolution_and_check_ssn
+        case async_state.status
+        when :none
+          enqueue_job
+        when :in_progress
+          nil
+        when :timed_out
+          enqueue_job
+        when :done
+          nil
+        end
+      end
+
+      def after_call(pii, idv_result)
+        # binding.pry
+
+        add_proofing_costs(idv_result)
+        response = idv_result_to_form_response(idv_result)
+        response = check_ssn(pii) if response.success?
+        summarize_result_and_throttle_failures(response)
+      end
+
+      def async?
+        true
+      end
+
+      # @return [State]
+      def async_state
+        dcs_uuid = flow_session[:idv_verify_step_document_capture_session_uuid]
+        dcs = DocumentCaptureSession.find_by(uuid: dcs_uuid)
+        return State.none if dcs_uuid.nil?
+        return State.timed_out if dcs.nil?
+
+        proofing_job_result = dcs.load_proofing_result
+        return State.timed_out if proofing_job_result.nil?
+
+        if proofing_job_result.result
+          proofing_job_result.result.deep_symbolize_keys!
+          proofing_job_result.pii.deep_symbolize_keys!
+          State.done(pii: proofing_job_result.pii, result: proofing_job_result.result)
+        elsif dcs.pii
+          State.in_progress
+        end
+      end
+
+      def delete_async
+        flow_session.delete(:idv_verify_step_document_capture_session_uuid)
+      end
+
+      private
+
+      def enqueue_job
+        pii_from_doc = flow_session[:pii_from_doc]
+
+        document_capture_session = DocumentCaptureSession.create(user_id: user_id,
+                                                                 requested_at: Time.zone.now)
+        document_capture_session.store_proofing_pii_from_doc(pii_from_doc)
+
+        flow_session[:idv_verify_step_document_capture_session_uuid] = document_capture_session.uuid
+
+        stages = should_use_aamva?(pii_from_doc) ? %w[resolution state_id] : ['resolution']
+        VendorProofJob.perform_later(document_capture_session.uuid, stages)
       end
     end
   end
diff --git a/app/services/proofing_document_capture_session_result.rb b/app/services/proofing_document_capture_session_result.rb
new file mode 100644
index 00000000000..d5cd829bb32
--- /dev/null
+++ b/app/services/proofing_document_capture_session_result.rb
@@ -0,0 +1,59 @@
+class ProofingDocumentCaptureSessionResult
+  REDIS_KEY_PREFIX = 'dcs-proofing:result'.freeze
+
+  attr_reader :id, :pii, :result
+
+  class << self
+    def load(id)
+      ciphertext = REDIS_POOL.with { |client| client.read(key(id)) }
+      return nil if ciphertext.blank?
+      decrypt_and_deserialize(id, ciphertext)
+    end
+
+    def store(id:, pii:, result:)
+      result = new(id: id, pii: pii, result: result)
+      REDIS_POOL.with do |client|
+        client.write(key(id), result.serialize_and_encrypt, expires_in: 60)
+      end
+    end
+
+    def key(id)
+      [REDIS_KEY_PREFIX, id].join(':')
+    end
+
+    private
+
+    def decrypt_and_deserialize(id, ciphertext)
+      deserialize(
+        id,
+        Encryption::Encryptors::SessionEncryptor.new.decrypt(ciphertext),
+      )
+    end
+
+    def deserialize(id, json)
+      data = JSON.parse(json)
+      new(
+        id: id,
+        pii: data['pii'],
+        result: data['result'],
+      )
+    end
+  end
+
+  def initialize(id:, pii:, result:)
+    @id = id
+    @pii = pii
+    @result = result
+  end
+
+  def serialize
+    {
+      pii: pii,
+      result: result,
+    }.to_json
+  end
+
+  def serialize_and_encrypt
+    Encryption::Encryptors::SessionEncryptor.new.encrypt(serialize)
+  end
+end
diff --git a/config/application.rb b/config/application.rb
index b16ce11fa1d..3b465806c45 100644
--- a/config/application.rb
+++ b/config/application.rb
@@ -8,7 +8,7 @@
 
 module Upaya
   class Application < Rails::Application
-    config.active_job.queue_adapter = 'inline'
+    config.active_job.queue_adapter = :inline
     config.autoload_paths << Rails.root.join('app', 'mailers', 'concerns')
     config.time_zone = 'UTC'
 
diff --git a/config/environments/test.rb b/config/environments/test.rb
index 1cfcf9d2460..da0482a1e7e 100644
--- a/config/environments/test.rb
+++ b/config/environments/test.rb
@@ -1,5 +1,5 @@
 Rails.application.configure do
-  config.active_job.queue_adapter = :test
+  config.active_job.queue_adapter = :inline
   config.cache_classes = true
   config.eager_load = false
   config.public_file_server.enabled = true
diff --git a/spec/config/initializers/active_job_logger_patch_spec.rb b/spec/config/initializers/active_job_logger_patch_spec.rb
index aade0319acb..9bef79f9aae 100644
--- a/spec/config/initializers/active_job_logger_patch_spec.rb
+++ b/spec/config/initializers/active_job_logger_patch_spec.rb
@@ -23,7 +23,7 @@ def perform(sensitive_param:); end
 
     # In this case, we need to assert before the action which logs, block-style to
     # match the initializer
-    expect(Rails.logger).to receive(:info) do |&blk|
+    expect(Rails.logger).to receive(:info).exactly(3).times do |&blk|
       output = JSON.parse(blk.call)
 
       # [Sidenote: The nested assertions don't seem to be reflected in the spec
diff --git a/spec/features/idv/steps/phone_otp_delivery_selection_step_spec.rb b/spec/features/idv/steps/phone_otp_delivery_selection_step_spec.rb
index 38a74bcb6c8..995520db634 100644
--- a/spec/features/idv/steps/phone_otp_delivery_selection_step_spec.rb
+++ b/spec/features/idv/steps/phone_otp_delivery_selection_step_spec.rb
@@ -1,6 +1,6 @@
 require 'rails_helper'
 
-feature 'IdV phone OTP deleivery method selection' do
+feature 'IdV phone OTP delivery method selection' do
   include IdvStepHelper
 
   context 'the users chooses sms' do
diff --git a/spec/jobs/vendor_proof_job_spec.rb b/spec/jobs/vendor_proof_job_spec.rb
new file mode 100644
index 00000000000..9e4e70817d0
--- /dev/null
+++ b/spec/jobs/vendor_proof_job_spec.rb
@@ -0,0 +1,5 @@
+require 'rails_helper'
+
+RSpec.describe VendorProofJob, type: :job do
+  pending "add some examples to (or delete) #{__FILE__}"
+end