diff --git a/app/controllers/account_statements_controller.rb b/app/controllers/account_statements_controller.rb
index 27793c4b6..3c3bd992a 100644
--- a/app/controllers/account_statements_controller.rb
+++ b/app/controllers/account_statements_controller.rb
@@ -1,8 +1,8 @@
 # frozen_string_literal: true
 
 class AccountStatementsController < ApplicationController
-  before_action :set_statement, only: %i[show update destroy link unlink reject]
-  before_action :ensure_statement_manager!, only: %i[index create update destroy link unlink reject]
+  before_action :set_statement, only: %i[show update destroy extract link unlink reject]
+  before_action :ensure_statement_manager!, only: %i[index create update destroy extract link unlink reject]
 
   def index
     accessible_account_ids = Current.user.accessible_accounts.select(:id)
@@ -30,6 +30,7 @@ def show
     @accounts = Current.user.accessible_accounts.visible.alphabetically
     @can_manage_statement = @statement.manageable_by?(Current.user)
     @reconciliation_checks = @statement.reconciliation_checks
+    @latest_pdf_import = @statement.latest_reusable_pdf_import if @statement.pdf?
     @breadcrumbs = [
       [ t("breadcrumbs.home"), root_path ],
       [ t("account_statements.index.title"), account_statements_path ],
@@ -41,6 +42,7 @@ def show
   def create
     files = Array(statement_upload_params[:files]).reject(&:blank?).select { |file| file.respond_to?(:read) }
     account = target_account
+    return if performed?
 
     if files.empty?
       redirect_back_or_to account_statements_path, alert: t("account_statements.create.no_files")
@@ -56,22 +58,24 @@ def create
     files.each do |file|
       prepared_upload = AccountStatement.prepare_upload!(file)
       created << AccountStatement.create_from_prepared_upload!(family: Current.family, account: account, prepared_upload: prepared_upload)
-    rescue AccountStatement::InvalidUploadError
-      validation_errors << t("account_statements.create.invalid_file_type")
+    rescue AccountStatement::InvalidUploadError => e
+      validation_errors << invalid_upload_message(e)
     rescue AccountStatement::DuplicateUploadError => e
       duplicates << e.statement
     rescue ActiveRecord::RecordInvalid => e
       validation_errors << e.record.errors.full_messages.to_sentence
     end
 
-    redirect_to redirect_after_create(account, created.first || duplicates.first),
+    redirect_statement = created.first || duplicates.find { |statement| statement.viewable_by?(Current.user) }
+    redirect_to redirect_after_create(account, redirect_statement),
                 flash_for_upload(created:, duplicates:, validation_errors:)
   end
 
   def update
     return if @statement.account && !require_account_permission!(@statement.account)
 
-    target = statement_account_id.present? ? Current.user.accessible_accounts.find(statement_account_id) : nil
+    target = statement_account_id.present? ? target_account : nil
+    return if performed?
     return if target && !require_account_permission!(target)
 
     attrs = statement_params.to_h
@@ -100,7 +104,11 @@ def link
       return
     end
 
-    account = Current.user.accessible_accounts.find(account_id)
+    account = Current.user.accessible_accounts.find_by(id: account_id)
+    unless account
+      redirect_to account_statement_path(@statement), alert: t("accounts.not_authorized")
+      return
+    end
     return unless require_account_permission!(account)
 
     @statement.link_to_account!(account)
@@ -116,6 +124,7 @@ def unlink
 
   def reject
     return if @statement.account && !require_account_permission!(@statement.account)
+    return redirect_to(account_statement_path(@statement), alert: t("account_statements.reject.linked")) if @statement.account.present?
 
     @statement.reject_match!
     redirect_to account_statements_path, notice: t("account_statements.reject.success")
@@ -132,6 +141,23 @@ def destroy
     end
   end
 
+  def extract
+    unless @statement.manageable_by?(Current.user)
+      redirect_to account_statement_path(@statement), alert: t("accounts.not_authorized")
+      return
+    end
+
+    unless @statement.pdf?
+      redirect_to account_statement_path(@statement), alert: t("account_statements.extract.not_pdf")
+      return
+    end
+
+    pdf_import = PdfImport.create_from_statement!(statement: @statement)
+    pdf_import.process_with_ai_later
+
+    redirect_to import_path(pdf_import), notice: t("account_statements.extract.started")
+  end
+
   private
 
     def set_statement
@@ -146,7 +172,7 @@ def set_statement
     def ensure_statement_manager!
       return if AccountStatement.statement_manager?(Current.user)
 
-      redirect_to accounts_path, alert: t("accounts.not_authorized")
+      redirect_back_or_to accounts_path, alert: t("accounts.not_authorized")
     end
 
     def statement_upload_params
@@ -170,7 +196,11 @@ def target_account
       account_id = statement_account_id.presence
       return nil if account_id.blank?
 
-      Current.user.accessible_accounts.find(account_id)
+      account = Current.user.accessible_accounts.find_by(id: account_id)
+      return account if account
+
+      redirect_back_or_to account_statements_path, alert: t("accounts.not_authorized")
+      nil
     end
 
     def statement_account_id
@@ -191,6 +221,14 @@ def redirect_after_create(account, statement = nil)
       end
     end
 
+    def invalid_upload_message(error)
+      reason = error.respond_to?(:reason) ? error.reason : nil
+      key = reason.in?(%i[oversize unsupported_type empty invalid_pdf_header]) ? reason : :invalid_file_type
+
+      t("account_statements.create.upload_errors.#{key}",
+        default: t("account_statements.create.invalid_file_type"))
+    end
+
     def post_link_path(statement)
       statement.account ? account_path(statement.account, tab: "statements") : account_statement_path(statement)
     end
diff --git a/app/controllers/imports_controller.rb b/app/controllers/imports_controller.rb
index 4a3cc0492..8ba5a2e06 100644
--- a/app/controllers/imports_controller.rb
+++ b/app/controllers/imports_controller.rb
@@ -2,9 +2,9 @@ class ImportsController < ApplicationController
   include SettingsHelper
 
   before_action :set_import, only: %i[show update publish destroy revert apply_template]
+  before_action :require_statement_import_permission!, only: %i[update publish destroy revert apply_template]
 
   def update
-    # Handle both pdf_import[account_id] and import[account_id] param formats
     account_id = params.dig(:pdf_import, :account_id) || params.dig(:import, :account_id)
 
     if account_id.present?
@@ -13,7 +13,9 @@ def update
         redirect_back_or_to import_path(@import), alert: t("imports.update.invalid_account", default: "Account not found.")
         return
       end
-      @import.update!(account: account)
+      return if @import.account_statement.present? && !require_account_permission!(account)
+
+      @import.assign_account!(account)
     end
 
     redirect_to import_path(@import), notice: t("imports.update.account_saved", default: "Account saved.")
@@ -54,12 +56,7 @@ def create
       return
     end
 
-    # Handle PDF file uploads - process with AI
     if file.present? && Import::ALLOWED_PDF_MIME_TYPES.include?(file.content_type)
-      unless valid_pdf_file?(file)
-        redirect_to new_import_path, alert: t("imports.create.invalid_pdf")
-        return
-      end
       create_pdf_import(file)
       return
     end
@@ -87,8 +84,6 @@ def create
         return
       end
 
-      # Stream reading is not fully applicable here as we store the raw string in the DB,
-      # but we have validated size beforehand to prevent memory exhaustion from massive files.
       import.update!(raw_file_str: file.read)
 
       redirect_to import_configuration_path(import), notice: t("imports.create.csv_uploaded")
@@ -132,24 +127,41 @@ def destroy
 
   private
     def set_import
-      @import = Current.family.imports.includes(:account).find(params[:id])
+      @import = Current.family.imports.includes(:account, :account_statement).find(params[:id])
+      raise ActiveRecord::RecordNotFound if @import.account_statement.present? && !@import.account_statement.viewable_by?(Current.user)
     end
 
     def import_params
       params.require(:import).permit(:import_file)
     end
 
+    def require_statement_import_permission!
+      return unless @import.account_statement.present?
+      return if @import.account_statement.manageable_by?(Current.user)
+
+      redirect_target = @import.account || @import.account_statement
+      redirect_back_or_to redirect_target, alert: t("accounts.not_authorized")
+    end
+
     def create_pdf_import(file)
+      unless AccountStatement.statement_manager?(Current.user)
+        redirect_to new_import_path, alert: t("accounts.not_authorized")
+        return
+      end
+
       if file.size > Import::MAX_PDF_SIZE
         redirect_to new_import_path, alert: t("imports.create.pdf_too_large", max_size: Import::MAX_PDF_SIZE / 1.megabyte)
         return
       end
 
-      pdf_import = Current.family.imports.create!(type: "PdfImport")
-      pdf_import.pdf_file.attach(file)
+      pdf_import = PdfImport.create_from_upload!(family: Current.family, file: file, user: Current.user)
       pdf_import.process_with_ai_later
 
       redirect_to import_path(pdf_import), notice: t("imports.create.pdf_processing")
+    rescue AccountStatement::DuplicateUploadError
+      redirect_to new_import_path, alert: t("imports.create.duplicate_pdf_unavailable")
+    rescue AccountStatement::InvalidUploadError
+      redirect_to new_import_path, alert: t("imports.create.invalid_pdf")
     end
 
     def create_document_import(file)
@@ -174,11 +186,6 @@ def create_document_import(file)
       end
 
       if ext == ".pdf"
-        unless valid_pdf_file?(file)
-          redirect_to new_import_path, alert: t("imports.create.invalid_pdf")
-          return
-        end
-
         create_pdf_import(file)
         return
       end
@@ -239,10 +246,4 @@ def create_sure_import(file)
 
       redirect_to import_path(import), notice: t("imports.create.ndjson_uploaded")
     end
-
-    def valid_pdf_file?(file)
-      header = file.read(5)
-      file.rewind
-      header&.start_with?("%PDF-")
-    end
 end
diff --git a/app/jobs/process_pdf_job.rb b/app/jobs/process_pdf_job.rb
index 1ed48cf49..0145b1541 100644
--- a/app/jobs/process_pdf_job.rb
+++ b/app/jobs/process_pdf_job.rb
@@ -1,11 +1,13 @@
 class ProcessPdfJob < ApplicationJob
+  PROCESSING_CLAIM_TTL = 30.minutes
+
   queue_as :medium_priority
 
   def perform(pdf_import)
     return unless pdf_import.is_a?(PdfImport)
-    return unless pdf_import.pdf_uploaded?
+    return reset_processing_claim(pdf_import) unless pdf_import.pdf_uploaded?
     return if pdf_import.status == "complete"
-    return if pdf_import.ai_processed? && (!pdf_import.statement_with_transactions? || pdf_import.rows_count > 0)
+    return reset_processing_claim(pdf_import) if pdf_import.ai_processed? && (!pdf_import.statement_with_transactions? || pdf_import.rows_count > 0)
 
     pdf_import.update!(status: :importing)
 
@@ -62,7 +64,7 @@ def sanitize_error_message(error)
     end
 
     def upload_to_vector_store(pdf_import, document_type:)
-      filename = pdf_import.pdf_file.filename.to_s
+      filename = pdf_import.pdf_filename
       file_content = pdf_import.pdf_file_content
 
       family_document = pdf_import.family.upload_document(
@@ -85,4 +87,16 @@ def resolve_document_type(pdf_import, process_result)
     def statement_with_transactions?(document_type)
       document_type.in?(%w[bank_statement credit_card_statement])
     end
+
+    def reset_processing_claim(pdf_import)
+      pdf_import.with_lock do
+        if pdf_import.importing? && processing_claim_stale?(pdf_import)
+          pdf_import.update!(status: :pending)
+        end
+      end
+    end
+
+    def processing_claim_stale?(pdf_import)
+      pdf_import.updated_at <= PROCESSING_CLAIM_TTL.ago
+    end
 end
diff --git a/app/models/account_statement.rb b/app/models/account_statement.rb
index 07cc80c86..4befd4664 100644
--- a/app/models/account_statement.rb
+++ b/app/models/account_statement.rb
@@ -15,7 +15,15 @@ def initialize(statement)
       super("Statement file has already been uploaded")
     end
   end
-  InvalidUploadError = Class.new(StandardError)
+  InvalidUploadError = Class.new(StandardError) do
+    attr_reader :reason
+
+    def initialize(reason = :invalid_file_type)
+      @reason = reason
+      super("Invalid statement upload: #{reason}")
+    end
+  end
+  LinkedStatementRejectionError = Class.new(StandardError)
 
   PreparedUpload = Data.define(:content, :filename, :content_type, :byte_size, :checksum, :content_sha256)
 
@@ -33,6 +41,7 @@ def initialize(statement)
   belongs_to :account, optional: true
   belongs_to :suggested_account, class_name: "Account", optional: true
 
+  has_many :pdf_imports, -> { where(type: "PdfImport").ordered }, class_name: "PdfImport", dependent: :nullify
   has_one_attached :original_file, dependent: :purge_later
 
   enum :source, { manual_upload: "manual_upload" }, validate: true, default: "manual_upload"
@@ -82,6 +91,7 @@ def create_from_upload!(family:, account:, file:)
 
     def create_from_prepared_upload!(family:, account:, prepared_upload:)
       statement = nil
+      saved = false
       duplicate = duplicate_for(family, prepared_upload)
       raise DuplicateUploadError, duplicate if duplicate
 
@@ -107,19 +117,15 @@ def create_from_prepared_upload!(family:, account:, prepared_upload:)
       MetadataDetector.new(statement, content: prepared_upload.content).apply
       statement.assign_account_match unless account.present?
       statement.save!
+      saved = true
       statement
     rescue ActiveRecord::RecordNotUnique
       duplicate = duplicate_for(family, prepared_upload)
-      purge_original_file(statement)
-
-      if duplicate
-        raise DuplicateUploadError, duplicate
-      end
+      raise DuplicateUploadError, duplicate if duplicate
 
       raise
-    rescue StandardError
-      purge_original_file(statement)
-      raise
+    ensure
+      purge_original_file(statement) if statement && !saved
     end
 
     def reconciliation_statuses_for(statements, account:)
@@ -135,11 +141,11 @@ def prepare_upload!(file)
       filename = file.original_filename.to_s
       content = read_upload_content!(file)
       byte_size = content.bytesize
-      raise InvalidUploadError if byte_size.zero?
+      raise InvalidUploadError, :empty if byte_size.zero?
 
       content_type = detected_content_type(content:, filename:, declared_content_type: file.content_type)
-      raise InvalidUploadError unless allowed_upload?(filename:, content_type:)
-      raise InvalidUploadError if content_type == "application/pdf" && !valid_pdf_content?(content)
+      raise InvalidUploadError, :unsupported_type unless allowed_upload?(filename:, content_type:)
+      raise InvalidUploadError, :invalid_pdf_header if content_type == "application/pdf" && !valid_pdf_content?(content)
 
       PreparedUpload.new(
         content: content,
@@ -194,7 +200,7 @@ def balance_lookup_for(account, statements)
 
     def read_upload_content!(file)
       declared_size = declared_upload_size(file)
-      raise InvalidUploadError if declared_size.present? && declared_size > MAX_FILE_SIZE
+      raise InvalidUploadError, :oversize if declared_size.present? && declared_size > MAX_FILE_SIZE
 
       content = +"".b
       loop do
@@ -202,7 +208,7 @@ def read_upload_content!(file)
         break if chunk.nil? || chunk.empty?
 
         content << chunk
-        raise InvalidUploadError if content.bytesize > MAX_FILE_SIZE
+        raise InvalidUploadError, :oversize if content.bytesize > MAX_FILE_SIZE
       end
 
       file.rewind if file.respond_to?(:rewind)
@@ -254,7 +260,7 @@ def link_to_account!(target_account, confidence: 1.0)
 
   def unlink!
     transaction do
-      update!(
+      assign_attributes(
         account: nil,
         review_status: :unmatched,
         match_confidence: nil
@@ -265,6 +271,8 @@ def unlink!
   end
 
   def reject_match!
+    raise LinkedStatementRejectionError, "Linked statements cannot reject a suggested match" if account.present?
+
     update!(
       suggested_account: nil,
       match_confidence: nil,
@@ -360,6 +368,10 @@ def xlsx?
     content_type.in?(ALLOWED_EXTENSION_CONTENT_TYPES[".xlsx"])
   end
 
+  def latest_reusable_pdf_import
+    pdf_imports.where.not(status: :failed).order(created_at: :desc).first
+  end
+
   private
 
     def reconciliation_check(key:, statement_amount:, ledger_amount:)
@@ -388,7 +400,7 @@ def sync_file_metadata
     end
 
     def normalize_currency
-      self.currency = currency.to_s.upcase.presence if currency.present?
+      self.currency = currency.to_s.strip.upcase.presence if currency.present?
     end
 
     def sync_review_status
diff --git a/app/models/account_statement/account_matcher.rb b/app/models/account_statement/account_matcher.rb
index 6e8c5c1c2..6ce84c893 100644
--- a/app/models/account_statement/account_matcher.rb
+++ b/app/models/account_statement/account_matcher.rb
@@ -10,7 +10,7 @@ def initialize(statement)
   end
 
   def best_match
-    candidates = statement.family.accounts.visible.to_a.filter_map do |account|
+    candidates = statement.family.accounts.visible.includes(:account_providers).to_a.filter_map do |account|
       confidence = confidence_for(account)
       next if confidence < 0.35
 
diff --git a/app/models/account_statement/coverage.rb b/app/models/account_statement/coverage.rb
index 225df183c..ef4c251e4 100644
--- a/app/models/account_statement/coverage.rb
+++ b/app/models/account_statement/coverage.rb
@@ -70,8 +70,8 @@ def default_expected_start_month(account, fallback_end_month: default_expected_e
       candidates = [
         account.entries.minimum(:date),
         account.balances.minimum(:date),
-        account.account_statements.where.not(period_start_on: nil).minimum(:period_start_on),
-        account.family.account_statements.unmatched.where(suggested_account: account).where.not(period_start_on: nil).minimum(:period_start_on)
+        account.account_statements.where.not(period_start_on: nil).where.not(period_end_on: nil).minimum(:period_start_on),
+        account.family.account_statements.unmatched.where(suggested_account: account).where.not(period_start_on: nil).where.not(period_end_on: nil).minimum(:period_start_on)
       ].compact
 
       start_month = (candidates.min || fallback_end_month.advance(months: -11)).to_date.beginning_of_month
diff --git a/app/models/account_statement/metadata_detector.rb b/app/models/account_statement/metadata_detector.rb
index e8079d5b8..05c488b8e 100644
--- a/app/models/account_statement/metadata_detector.rb
+++ b/app/models/account_statement/metadata_detector.rb
@@ -115,7 +115,6 @@ def detect_from_filename
 
       if (meaningful_hint = meaningful_filename_hint(hint))
         statement.institution_name_hint ||= meaningful_hint
-        statement.account_name_hint ||= meaningful_hint
         detected = true
       end
 
diff --git a/app/models/import.rb b/app/models/import.rb
index 2f1256ff4..fd1b63ce3 100644
--- a/app/models/import.rb
+++ b/app/models/import.rb
@@ -29,10 +29,15 @@ def self.max_csv_size
     MAX_CSV_SIZE
   end
 
+  def assign_account!(account)
+    update!(account: account)
+  end
+
   AMOUNT_TYPE_STRATEGIES = %w[signed_amount custom_column].freeze
 
   belongs_to :family
   belongs_to :account, optional: true
+  belongs_to :account_statement, optional: true
 
   before_validation :set_default_number_format
   before_validation :ensure_utf8_encoding
diff --git a/app/models/pdf_import.rb b/app/models/pdf_import.rb
index f610e6b0f..b80725793 100644
--- a/app/models/pdf_import.rb
+++ b/app/models/pdf_import.rb
@@ -2,6 +2,46 @@ class PdfImport < Import
   has_one_attached :pdf_file, dependent: :purge_later
 
   validates :document_type, inclusion: { in: DOCUMENT_TYPES }, allow_nil: true
+  validate :account_statement_belongs_to_family
+  validate :account_statement_is_pdf
+
+  class << self
+    def create_from_upload!(family:, file:, user:)
+      prepared_upload = AccountStatement.prepare_upload!(file)
+      statement = AccountStatement.create_from_prepared_upload!(
+        family: family,
+        account: nil,
+        prepared_upload: prepared_upload
+      )
+
+      create_from_statement!(statement: statement)
+    rescue AccountStatement::DuplicateUploadError => e
+      raise unless e.statement.viewable_by?(user)
+
+      create_from_statement!(statement: e.statement)
+    end
+
+    def create_from_statement!(statement:)
+      reusable_import = statement.latest_reusable_pdf_import
+      return reusable_import if reusable_pdf_import_current?(reusable_import, statement)
+
+      create!(
+        family: statement.family,
+        account: statement.account,
+        account_statement: statement,
+        date_format: statement.family.date_format,
+        status: :pending
+      )
+    end
+
+    private
+
+      def reusable_pdf_import_current?(pdf_import, statement)
+        pdf_import.present? &&
+          pdf_import.account_id == statement.account_id &&
+          pdf_import.date_format == statement.family.date_format
+      end
+  end
 
   def import!
     raise "Account required for PDF import" unless account.present?
@@ -31,8 +71,19 @@ def import!
     end
   end
 
+  def assign_account!(account)
+    transaction do
+      update!(account: account)
+      statement = account_statement
+      if statement.present? && account.present?
+        statement.lock!
+        statement.link_to_account!(account) if statement.account_id != account.id
+      end
+    end
+  end
+
   def pdf_uploaded?
-    pdf_file.attached?
+    source_pdf_attached?
   end
 
   def ai_processed?
@@ -40,7 +91,25 @@ def ai_processed?
   end
 
   def process_with_ai_later
-    ProcessPdfJob.perform_later(self)
+    should_enqueue = with_lock do
+      if pending? && !ai_processed? && rows_count.zero?
+        update!(status: :importing)
+        true
+      else
+        false
+      end
+    end
+
+    return false unless should_enqueue
+
+    begin
+      ProcessPdfJob.perform_later(self)
+      true
+    rescue StandardError => e
+      Rails.logger.error("Failed to enqueue PDF processing for import #{id}: #{e.class.name} - #{e.message}")
+      reload.with_lock { update!(status: :pending) }
+      false
+    end
   end
 
   def process_with_ai
@@ -172,9 +241,29 @@ def requires_csv_workflow?
   end
 
   def pdf_file_content
-    return nil unless pdf_file.attached?
+    return @pdf_file_content if defined?(@pdf_file_content)
 
-    pdf_file.download
+    @pdf_file_content = if statement_backed?
+      account_statement.original_file.download
+    elsif pdf_file.attached?
+      pdf_file.download
+    end
+  end
+
+  def pdf_filename
+    if statement_backed?
+      account_statement.filename
+    elsif pdf_file.attached?
+      pdf_file.filename.to_s
+    end
+  end
+
+  def statement_backed?
+    account_statement&.original_file&.attached?
+  end
+
+  def legacy_pdf_file?
+    pdf_file.attached?
   end
 
   def required_column_keys
@@ -199,4 +288,22 @@ def format_date_for_import(date_str)
     rescue ArgumentError
       date_str.to_s
     end
+
+    def source_pdf_attached?
+      statement_backed? || legacy_pdf_file?
+    end
+
+    def account_statement_belongs_to_family
+      return if account_statement.blank?
+      return if account_statement.family_id == family_id
+
+      errors.add(:account_statement, :invalid)
+    end
+
+    def account_statement_is_pdf
+      return if account_statement.blank?
+      return if account_statement.pdf?
+
+      errors.add(:account_statement, :invalid)
+    end
 end
diff --git a/app/views/account_statements/show.html.erb b/app/views/account_statements/show.html.erb
index e320fc79a..69ae21674 100644
--- a/app/views/account_statements/show.html.erb
+++ b/app/views/account_statements/show.html.erb
@@ -138,6 +138,32 @@
           <% end %>
         </div>
 
+        <div class="rounded-xl bg-container-inset p-4">
+          <h3 class="text-sm font-medium text-primary mb-3"><%= t("account_statements.extract.title") %></h3>
+
+          <% if @statement.pdf? %>
+            <div class="space-y-3">
+              <% if @latest_pdf_import.present? %>
+                <%= link_to import_path(@latest_pdf_import), class: "inline-flex items-center gap-2 text-sm font-medium text-primary hover:text-primary-hover" do %>
+                  <%= icon("file-search", size: "sm") %>
+                  <%= t("account_statements.extract.latest_import") %>
+                <% end %>
+              <% end %>
+
+              <% if @can_manage_statement %>
+                <%= button_to extract_account_statement_path(@statement),
+                              method: :post,
+                              class: "inline-flex items-center gap-2 text-sm font-medium text-primary" do %>
+                  <%= icon("scan-text", size: "sm") %>
+                  <%= t("account_statements.extract.start") %>
+                <% end %>
+              <% end %>
+            </div>
+          <% else %>
+            <p class="text-sm text-secondary"><%= t("account_statements.extract.unavailable") %></p>
+          <% end %>
+        </div>
+
         <div class="rounded-xl bg-container-inset p-4">
           <h3 class="text-sm font-medium text-primary mb-3"><%= t(".reconciliation_title") %></h3>
 
diff --git a/app/views/accounts/show/_menu.html.erb b/app/views/accounts/show/_menu.html.erb
index cb577369e..99d5a29fe 100644
--- a/app/views/accounts/show/_menu.html.erb
+++ b/app/views/accounts/show/_menu.html.erb
@@ -6,7 +6,9 @@
     <% menu.with_item(variant: "link", text: "Edit", href: edit_account_path(account), icon: "pencil-line", data: { turbo_frame: :modal }) %>
   <% end %>
   <% menu.with_item(variant: "link", text: "Sharing", href: account_sharing_path(account), icon: "users", data: { turbo_frame: :modal }) %>
-  <% menu.with_item(variant: "link", text: t(".statements"), href: account_path(account, tab: "statements"), icon: "archive") %>
+  <% if permission.in?([ :owner, :full_control ]) && AccountStatement.statement_manager?(Current.user) %>
+    <% menu.with_item(variant: "link", text: t(".statements"), href: account_path(account, tab: "statements"), icon: "archive") %>
+  <% end %>
 
   <% if permission.in?([ :owner, :full_control ]) %>
     <% if account.supports_trades? %>
diff --git a/app/views/imports/_pdf_import.html.erb b/app/views/imports/_pdf_import.html.erb
index 068d55da9..097087a7b 100644
--- a/app/views/imports/_pdf_import.html.erb
+++ b/app/views/imports/_pdf_import.html.erb
@@ -14,16 +14,25 @@
       </div>
 
       <div class="bg-container border border-primary rounded-xl p-4 space-y-4">
+        <% if import.account_statement.present? %>
+          <div class="space-y-2">
+            <h2 class="font-medium text-primary"><%= t("imports.pdf_import.source_statement") %></h2>
+            <p class="text-sm text-secondary px-3 py-2 bg-container-inset rounded-lg">
+              <%= link_to import.account_statement.filename, account_statement_path(import.account_statement), class: "text-primary hover:text-primary-hover" %>
+            </p>
+          </div>
+        <% end %>
+
         <div class="space-y-2">
           <h2 class="font-medium text-primary"><%= t("imports.pdf_import.document_type_label") %></h2>
-          <p class="text-sm text-secondary px-3 py-2 bg-gray-tint-5 rounded-lg">
+          <p class="text-sm text-secondary px-3 py-2 bg-container-inset rounded-lg">
             <%= t("imports.document_types.#{import.document_type}", default: import.document_type&.humanize || t("imports.pdf_import.unknown_document_type", default: "Unknown")) %>
           </p>
         </div>
 
         <div class="space-y-2">
           <h2 class="font-medium text-primary"><%= t("imports.pdf_import.transactions_extracted", default: "Transactions Extracted") %></h2>
-          <p class="text-sm text-secondary px-3 py-2 bg-gray-tint-5 rounded-lg">
+          <p class="text-sm text-secondary px-3 py-2 bg-container-inset rounded-lg">
             <%= t("imports.pdf_import.transactions_extracted_count", count: import.rows_count, default: "%{count} transactions") %>
           </p>
         </div>
@@ -106,16 +115,25 @@
       </div>
 
       <div class="bg-container border border-primary rounded-xl p-4 space-y-4">
+        <% if import.account_statement.present? %>
+          <div class="space-y-2">
+            <h2 class="font-medium text-primary"><%= t("imports.pdf_import.source_statement") %></h2>
+            <p class="text-sm text-secondary px-3 py-2 bg-container-inset rounded-lg">
+              <%= link_to import.account_statement.filename, account_statement_path(import.account_statement), class: "text-primary hover:text-primary-hover" %>
+            </p>
+          </div>
+        <% end %>
+
         <div class="space-y-2">
           <h2 class="font-medium text-primary"><%= t("imports.pdf_import.document_type_label") %></h2>
-          <p class="text-sm text-secondary px-3 py-2 bg-gray-tint-5 rounded-lg">
+          <p class="text-sm text-secondary px-3 py-2 bg-container-inset rounded-lg">
             <%= t("imports.document_types.#{import.document_type}", default: import.document_type&.humanize || t("imports.pdf_import.unknown_document_type", default: "Unknown")) %>
           </p>
         </div>
 
         <div class="space-y-2">
           <h2 class="font-medium text-primary"><%= t("imports.pdf_import.summary_label") %></h2>
-          <p class="text-sm text-secondary px-3 py-2 bg-gray-tint-5 rounded-lg whitespace-pre-wrap">
+          <p class="text-sm text-secondary px-3 py-2 bg-container-inset rounded-lg whitespace-pre-wrap">
             <%= import.ai_summary %>
           </p>
         </div>
diff --git a/app/views/settings/_settings_nav.html.erb b/app/views/settings/_settings_nav.html.erb
index 19bd4a241..a48dd4019 100644
--- a/app/views/settings/_settings_nav.html.erb
+++ b/app/views/settings/_settings_nav.html.erb
@@ -61,7 +61,7 @@ nav_sections = [
       <kbd>esc</kbd>
     <% end %>
   </div>
-  <nav class="space-y-4 hidden md:block">
+  <nav class="space-y-4 hidden md:block" data-testid="settings-nav">
     <% nav_sections.compact.each do |section| %>
       <section class="space-y-2">
         <div class="flex items-center gap-2 px-3">
@@ -85,7 +85,7 @@ nav_sections = [
       <% end %>
     </section>
   </nav>
-  <nav class="space-y-4 overflow-x-auto md:hidden no-scrollbar overscroll-none" id="mobile-settings-nav" data-controller="preserve-scroll scroll-on-connect">
+  <nav class="space-y-4 overflow-x-auto md:hidden no-scrollbar overscroll-none" id="mobile-settings-nav" data-testid="settings-nav" data-controller="preserve-scroll scroll-on-connect">
     <ul class="flex flex-nowrap space-x-1">
       <li>
         <%= render DS::Link.new(
diff --git a/config/initializers/active_storage_authorization.rb b/config/initializers/active_storage_authorization.rb
index c856faa08..d56f3ac88 100644
--- a/config/initializers/active_storage_authorization.rb
+++ b/config/initializers/active_storage_authorization.rb
@@ -56,7 +56,7 @@ def account_statement_attachment_authorized?(attachment)
       end
 
       def authorized_attachments
-        return nil unless authorized_blob
+        return [] unless authorized_blob
 
         @authorized_attachments ||= ActiveStorage::Attachment.where(blob: authorized_blob).to_a
       end
diff --git a/config/locales/views/account_statements/en.yml b/config/locales/views/account_statements/en.yml
index 78e21ae57..dc6bf98b0 100644
--- a/config/locales/views/account_statements/en.yml
+++ b/config/locales/views/account_statements/en.yml
@@ -28,9 +28,22 @@ en:
       success:
         one: 1 statement uploaded.
         other: "%{count} statements uploaded."
+      upload_errors:
+        empty: The statement file is empty.
+        invalid_file_type: Upload a PDF, CSV, or XLSX statement under the size limit.
+        invalid_pdf_header: The PDF file does not look valid.
+        oversize: The statement file is too large.
+        unsupported_type: Upload a PDF, CSV, or XLSX statement under the size limit.
     destroy:
       failure: Statement could not be deleted.
       success: Statement deleted.
+    extract:
+      latest_import: Latest import
+      not_pdf: Transaction extraction is only available for PDF statements.
+      start: Extract transactions
+      started: Statement extraction started.
+      title: Transaction extraction
+      unavailable: Transaction extraction is available for PDF statements.
     form:
       account_upload: Upload statement
       files_hint: PDF, CSV, or XLSX. %{max_size}MB max per file.
@@ -64,6 +77,7 @@ en:
       mismatched: Mismatched
       unavailable: Not checked
     reject:
+      linked: Unlink the statement before rejecting a suggested match.
       success: Statement match rejected.
     show:
       account_label: Account
diff --git a/config/locales/views/imports/en.yml b/config/locales/views/imports/en.yml
index 9f41ab712..377321d21 100644
--- a/config/locales/views/imports/en.yml
+++ b/config/locales/views/imports/en.yml
@@ -207,6 +207,7 @@ en:
       pdf_too_large: PDF file is too large. Maximum size is %{max_size}MB.
       pdf_processing: Your PDF is being processed. You will receive an email when analysis is complete.
       invalid_pdf: The uploaded file is not a valid PDF.
+      duplicate_pdf_unavailable: This PDF is already stored as a statement you cannot access.
       document_too_large: Document file is too large. Maximum size is %{max_size}MB.
       invalid_document_file_type: Invalid document file type for the active vector store.
       document_uploaded: Document uploaded successfully.
@@ -248,6 +249,7 @@ en:
       complete_title: Document analyzed
       complete_description: We've analyzed your PDF and here's what we found.
       document_type_label: Document Type
+      source_statement: Source statement
       summary_label: Summary
       email_sent_notice: An email has been sent to you with next steps.
       back_to_imports: Back to imports
diff --git a/config/routes.rb b/config/routes.rb
index 5f4aa337a..c01b29221 100644
--- a/config/routes.rb
+++ b/config/routes.rb
@@ -437,6 +437,7 @@
 
   resources :account_statements, only: %i[index show create update destroy] do
     member do
+      post :extract
       patch :link
       patch :unlink
       patch :reject
diff --git a/db/migrate/20260513120000_add_account_statement_to_imports.rb b/db/migrate/20260513120000_add_account_statement_to_imports.rb
new file mode 100644
index 000000000..f46336e21
--- /dev/null
+++ b/db/migrate/20260513120000_add_account_statement_to_imports.rb
@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+class AddAccountStatementToImports < ActiveRecord::Migration[7.2]
+  def change
+    add_reference :imports,
+                  :account_statement,
+                  type: :uuid,
+                  foreign_key: { on_delete: :nullify },
+                  index: true
+  end
+end
diff --git a/db/schema.rb b/db/schema.rb
index 943e4dfe3..590cf7524 100644
--- a/db/schema.rb
+++ b/db/schema.rb
@@ -10,7 +10,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema[7.2].define(version: 2026_05_12_211200) do
+ActiveRecord::Schema[7.2].define(version: 2026_05_13_120000) do
   # These are extensions that must be enabled in order to support this database
   enable_extension "pgcrypto"
   enable_extension "plpgsql"
@@ -894,6 +894,8 @@
     t.text "ai_summary"
     t.string "document_type"
     t.jsonb "extracted_data"
+    t.uuid "account_statement_id"
+    t.index ["account_statement_id"], name: "index_imports_on_account_statement_id"
     t.index ["family_id"], name: "index_imports_on_family_id"
   end
 
@@ -1897,6 +1899,7 @@
   add_foreign_key "impersonation_sessions", "users", column: "impersonated_id"
   add_foreign_key "impersonation_sessions", "users", column: "impersonator_id"
   add_foreign_key "import_rows", "imports"
+  add_foreign_key "imports", "account_statements", on_delete: :nullify
   add_foreign_key "imports", "families"
   add_foreign_key "indexa_capital_accounts", "indexa_capital_items"
   add_foreign_key "indexa_capital_items", "families"
diff --git a/test/controllers/account_statements_controller_test.rb b/test/controllers/account_statements_controller_test.rb
index 01739f0dc..0f5806d10 100644
--- a/test/controllers/account_statements_controller_test.rb
+++ b/test/controllers/account_statements_controller_test.rb
@@ -44,6 +44,15 @@ class AccountStatementsControllerTest < ActionDispatch::IntegrationTest
     assert_equal I18n.t("accounts.not_authorized"), flash[:alert]
   end
 
+  test "non manager returns to previous page when blocked from statement vault" do
+    sign_in family_guest
+
+    get account_statements_url, headers: { "HTTP_REFERER" => root_url }
+
+    assert_redirected_to root_url
+    assert_equal I18n.t("accounts.not_authorized"), flash[:alert]
+  end
+
   test "non manager cannot view unmatched statement" do
     statement = AccountStatement.create_from_upload!(
       family: @account.family,
@@ -107,6 +116,83 @@ class AccountStatementsControllerTest < ActionDispatch::IntegrationTest
     assert_redirected_to account_statement_url(statement)
   end
 
+  test "invalid target account with no files redirects once" do
+    assert_no_difference "AccountStatement.count" do
+      post account_statements_url, params: {
+        account_statement: {
+          account_id: "0"
+        }
+      }
+    end
+
+    assert_redirected_to account_statements_url
+    assert_equal I18n.t("accounts.not_authorized"), flash[:alert]
+  end
+
+  test "starts pdf extraction from statement" do
+    statement = AccountStatement.create_from_upload!(
+      family: @account.family,
+      account: @account,
+      file: uploaded_file(
+        filename: "statement.pdf",
+        content_type: "application/pdf",
+        content: file_fixture("imports/sample_bank_statement.pdf").binread
+      )
+    )
+
+    assert_difference "Import.where(type: 'PdfImport').count", 1 do
+      post extract_account_statement_url(statement)
+    end
+
+    pdf_import = PdfImport.order(:created_at).last
+    assert_equal statement, pdf_import.account_statement
+    assert_redirected_to import_url(pdf_import)
+    assert_equal I18n.t("account_statements.extract.started"), flash[:notice]
+  end
+
+  test "statement detail links latest reusable pdf import" do
+    statement = AccountStatement.create_from_upload!(
+      family: @account.family,
+      account: @account,
+      file: uploaded_file(
+        filename: "statement.pdf",
+        content_type: "application/pdf",
+        content: file_fixture("imports/sample_bank_statement.pdf").binread
+      )
+    )
+    reusable_import = PdfImport.create_from_statement!(statement: statement)
+    failed_import = PdfImport.create!(
+      family: statement.family,
+      account: statement.account,
+      account_statement: statement,
+      date_format: statement.family.date_format,
+      status: :failed,
+      created_at: 1.minute.from_now,
+      updated_at: 1.minute.from_now
+    )
+
+    get account_statement_url(statement)
+
+    assert_response :success
+    assert_select "a[href='#{import_path(reusable_import)}']", text: I18n.t("account_statements.extract.latest_import")
+    assert_select "a[href='#{import_path(failed_import)}']", count: 0
+  end
+
+  test "does not start extraction for non pdf statement" do
+    statement = AccountStatement.create_from_upload!(
+      family: @account.family,
+      account: @account,
+      file: uploaded_file(filename: "statement.csv", content_type: "text/csv")
+    )
+
+    assert_no_difference "Import.count" do
+      post extract_account_statement_url(statement)
+    end
+
+    assert_redirected_to account_statement_url(statement)
+    assert_equal I18n.t("account_statements.extract.not_pdf"), flash[:alert]
+  end
+
   test "skips duplicate statement upload" do
     AccountStatement.create_from_upload!(
       family: @account.family,
@@ -127,6 +213,27 @@ class AccountStatementsControllerTest < ActionDispatch::IntegrationTest
     assert_equal I18n.t("account_statements.create.duplicates", count: 1), flash[:alert]
   end
 
+  test "duplicate upload for inaccessible statement redirects to vault" do
+    private_account = accounts(:investment)
+    AccountStatement.create_from_upload!(
+      family: private_account.family,
+      account: private_account,
+      file: uploaded_file(filename: "private_statement.csv", content_type: "text/csv", content: "date,amount\n2024-01-01,1\n")
+    )
+    sign_in users(:family_member)
+
+    assert_no_difference "AccountStatement.count" do
+      post account_statements_url, params: {
+        account_statement: {
+          files: [ uploaded_file(filename: "duplicate.csv", content_type: "text/csv", content: "date,amount\n2024-01-01,1\n") ]
+        }
+      }
+    end
+
+    assert_redirected_to account_statements_url
+    assert_equal I18n.t("account_statements.create.duplicates", count: 1), flash[:alert]
+  end
+
   test "continues upload loop after a validation error" do
     invalid_record = AccountStatement.new
     invalid_record.errors.add(:filename, "is invalid")
@@ -167,7 +274,7 @@ class AccountStatementsControllerTest < ActionDispatch::IntegrationTest
     end
 
     assert_redirected_to account_statements_url
-    assert_equal I18n.t("account_statements.create.invalid_file_type"), flash[:alert]
+    assert_equal I18n.t("account_statements.create.upload_errors.unsupported_type"), flash[:alert]
   end
 
   test "continues upload loop after an invalid file type" do
@@ -185,7 +292,7 @@ class AccountStatementsControllerTest < ActionDispatch::IntegrationTest
     statement = AccountStatement.order(:created_at).last
     assert_redirected_to account_statement_url(statement)
     assert_equal I18n.t("account_statements.create.success", count: 1), flash[:notice]
-    assert_includes flash[:alert], I18n.t("account_statements.create.invalid_file_type")
+    assert_includes flash[:alert], I18n.t("account_statements.create.upload_errors.unsupported_type")
   end
 
   test "rejects txt and xls statement uploads" do
@@ -202,7 +309,7 @@ class AccountStatementsControllerTest < ActionDispatch::IntegrationTest
       end
 
       assert_redirected_to account_statements_url
-      assert_equal I18n.t("account_statements.create.invalid_file_type"), flash[:alert]
+      assert_equal I18n.t("account_statements.create.upload_errors.unsupported_type"), flash[:alert]
     end
   end
 
@@ -224,7 +331,7 @@ class AccountStatementsControllerTest < ActionDispatch::IntegrationTest
       end
 
       assert_redirected_to account_statements_url
-      assert_equal I18n.t("account_statements.create.invalid_file_type"), flash[:alert]
+      assert_equal I18n.t("account_statements.create.upload_errors.empty"), flash[:alert]
     end
   end
 
@@ -251,7 +358,7 @@ class AccountStatementsControllerTest < ActionDispatch::IntegrationTest
     end
 
     assert_redirected_to account_statements_url
-    assert_equal I18n.t("account_statements.create.invalid_file_type"), flash[:alert]
+    assert_equal I18n.t("account_statements.create.upload_errors.oversize"), flash[:alert]
   end
 
   test "rejects cross-family account id" do
@@ -272,7 +379,8 @@ class AccountStatementsControllerTest < ActionDispatch::IntegrationTest
         }
       }
     end
-    assert_response :not_found
+    assert_redirected_to account_statements_url
+    assert_equal I18n.t("accounts.not_authorized"), flash[:alert]
   end
 
   test "read only shared user cannot upload to account" do
@@ -407,6 +515,22 @@ class AccountStatementsControllerTest < ActionDispatch::IntegrationTest
     assert_nil statement.suggested_account
   end
 
+  test "does not reject already linked statement" do
+    statement = AccountStatement.create_from_upload!(
+      family: @account.family,
+      account: @account,
+      file: uploaded_file(filename: "statement.csv", content_type: "text/csv", content: "date,amount\n2024-01-01,1\n")
+    )
+
+    patch reject_account_statement_url(statement)
+
+    assert_redirected_to account_statement_url(statement)
+    assert_equal I18n.t("account_statements.reject.linked"), flash[:alert]
+    statement.reload
+    assert statement.linked?
+    assert_equal @account, statement.account
+  end
+
   test "updates metadata" do
     statement = AccountStatement.create_from_upload!(
       family: @account.family,
@@ -419,7 +543,7 @@ class AccountStatementsControllerTest < ActionDispatch::IntegrationTest
         period_start_on: "2024-01-01",
         period_end_on: "2024-01-31",
         closing_balance: "123.45",
-        currency: "usd"
+        currency: " usd "
       }
     }
 
@@ -451,6 +575,26 @@ class AccountStatementsControllerTest < ActionDispatch::IntegrationTest
     assert statement.linked?
   end
 
+  test "metadata update handles invalid target account with redirect" do
+    statement = AccountStatement.create_from_upload!(
+      family: @account.family,
+      account: nil,
+      file: uploaded_file(filename: "statement.csv", content_type: "text/csv", content: "date,amount\n2024-01-01,1\n")
+    )
+    sign_in users(:family_member)
+
+    patch account_statement_url(statement), params: {
+      account_statement: {
+        account_id: accounts(:investment).id,
+        period_start_on: "2024-01-01"
+      }
+    }
+
+    assert_redirected_to account_statements_url
+    assert_equal I18n.t("accounts.not_authorized"), flash[:alert]
+    assert_nil statement.reload.account
+  end
+
   test "deletes statement" do
     statement = AccountStatement.create_from_upload!(
       family: @account.family,
diff --git a/test/controllers/accounts_controller_test.rb b/test/controllers/accounts_controller_test.rb
index 4b7604c3e..46dbf80d4 100644
--- a/test/controllers/accounts_controller_test.rb
+++ b/test/controllers/accounts_controller_test.rb
@@ -96,6 +96,25 @@ class AccountsControllerTest < ActionDispatch::IntegrationTest
     assert_select "input[type=file]", count: 0
   end
 
+  test "account menu hides statement link without statement management permission" do
+    account = accounts(:credit_card)
+    sign_in users(:family_member)
+
+    get account_url(account)
+
+    assert_response :success
+    statements_path = account_path(account, tab: "statements")
+    assert_select "[data-testid='account-menu'] a[href='#{statements_path}']", count: 0
+  end
+
+  test "account menu shows statement link for statement managers with write access" do
+    get account_url(@account)
+
+    assert_response :success
+    statements_path = account_path(@account, tab: "statements")
+    assert_select "[data-testid='account-menu'] a[href='#{statements_path}']", count: 1
+  end
+
   test "account activity marks trade amounts as privacy-sensitive" do
     trade_entry = entries(:trade)
     expected_amount = ApplicationController.helpers.format_money(-trade_entry.amount_money)
diff --git a/test/controllers/imports_controller_test.rb b/test/controllers/imports_controller_test.rb
index dbea78d2d..397aa0009 100644
--- a/test/controllers/imports_controller_test.rb
+++ b/test/controllers/imports_controller_test.rb
@@ -1,6 +1,8 @@
 require "test_helper"
 
 class ImportsControllerTest < ActionDispatch::IntegrationTest
+  include ActiveJob::TestHelper
+
   setup do
     sign_in @user = users(:family_admin)
   end
@@ -83,20 +85,217 @@ class ImportsControllerTest < ActionDispatch::IntegrationTest
     @user.family.expects(:upload_document).never
 
     assert_difference "Import.count", 1 do
-      post imports_url, params: {
-        import: {
-          type: "DocumentImport",
-          import_file: file_fixture_upload("imports/sample_bank_statement.pdf", "application/pdf")
+      assert_difference "AccountStatement.count", 1 do
+        post imports_url, params: {
+          import: {
+            type: "DocumentImport",
+            import_file: file_fixture_upload("imports/sample_bank_statement.pdf", "application/pdf")
+          }
         }
-      }
+      end
     end
 
     created_import = Import.order(:created_at).last
     assert_equal "PdfImport", created_import.type
+    assert_equal AccountStatement.order(:created_at).last, created_import.account_statement
+    assert_not created_import.pdf_file.attached?
+    assert_redirected_to import_url(created_import)
+    assert_equal I18n.t("imports.create.pdf_processing"), flash[:notice]
+  end
+
+  test "uploads pdf import through account statement" do
+    assert_difference "AccountStatement.count", 1 do
+      assert_difference "Import.where(type: 'PdfImport').count", 1 do
+        post imports_url, params: {
+          import: {
+            import_file: file_fixture_upload("imports/sample_bank_statement.pdf", "application/pdf")
+          }
+        }
+      end
+    end
+
+    statement = AccountStatement.order(:created_at).last
+    created_import = PdfImport.order(:created_at).last
+    assert_equal statement, created_import.account_statement
+    assert_not created_import.pdf_file.attached?
     assert_redirected_to import_url(created_import)
     assert_equal I18n.t("imports.create.pdf_processing"), flash[:notice]
   end
 
+  test "guest cannot create statement backed pdf import" do
+    sign_in users(:intro_user)
+
+    assert_no_difference [ "AccountStatement.count", "Import.where(type: 'PdfImport').count" ] do
+      assert_no_enqueued_jobs only: ProcessPdfJob do
+        post imports_url, params: {
+          import: {
+            import_file: file_fixture_upload("imports/sample_bank_statement.pdf", "application/pdf")
+          }
+        }
+      end
+    end
+
+    assert_redirected_to new_import_url
+    assert_equal I18n.t("accounts.not_authorized"), flash[:alert]
+  end
+
+  test "duplicate pdf import reuses account statement" do
+    statement = AccountStatement.create_from_upload!(
+      family: @user.family,
+      account: nil,
+      file: uploaded_file(
+        filename: "existing_statement.pdf",
+        content_type: "application/pdf",
+        content: file_fixture("imports/sample_bank_statement.pdf").binread
+      )
+    )
+
+    assert_no_difference "AccountStatement.count" do
+      assert_difference "Import.where(type: 'PdfImport').count", 1 do
+        post imports_url, params: {
+          import: {
+            import_file: file_fixture_upload("imports/sample_bank_statement.pdf", "application/pdf")
+          }
+        }
+      end
+    end
+
+    created_import = PdfImport.order(:created_at).last
+    assert_equal statement, created_import.account_statement
+    assert_redirected_to import_url(created_import)
+  end
+
+  test "duplicate pdf import does not enqueue processing twice for reused import" do
+    assert_difference "AccountStatement.count", 1 do
+      assert_difference "Import.where(type: 'PdfImport').count", 1 do
+        assert_enqueued_jobs 1, only: ProcessPdfJob do
+          post imports_url, params: {
+            import: {
+              import_file: file_fixture_upload("imports/sample_bank_statement.pdf", "application/pdf")
+            }
+          }
+
+          post imports_url, params: {
+            import: {
+              import_file: file_fixture_upload("imports/sample_bank_statement.pdf", "application/pdf")
+            }
+          }
+        end
+      end
+    end
+
+    created_import = PdfImport.order(:created_at).last
+    assert_equal "importing", created_import.status
+    assert_redirected_to import_url(created_import)
+  end
+
+  test "duplicate pdf import for inaccessible statement does not create import" do
+    AccountStatement.create_from_upload!(
+      family: @user.family,
+      account: accounts(:investment),
+      file: uploaded_file(
+        filename: "existing_statement.pdf",
+        content_type: "application/pdf",
+        content: file_fixture("imports/sample_bank_statement.pdf").binread
+      )
+    )
+
+    sign_in users(:family_member)
+
+    assert_no_difference [ "AccountStatement.count", "Import.where(type: 'PdfImport').count" ] do
+      post imports_url, params: {
+        import: {
+          import_file: file_fixture_upload("imports/sample_bank_statement.pdf", "application/pdf")
+        }
+      }
+    end
+
+    assert_redirected_to new_import_url
+    assert_equal I18n.t("imports.create.duplicate_pdf_unavailable"), flash[:alert]
+  end
+
+  test "setting statement backed pdf import account links source statement" do
+    statement = AccountStatement.create_from_upload!(
+      family: @user.family,
+      account: nil,
+      file: uploaded_file(
+        filename: "statement.pdf",
+        content_type: "application/pdf",
+        content: file_fixture("imports/sample_bank_statement.pdf").binread
+      )
+    )
+    pdf_import = PdfImport.create_from_statement!(statement: statement)
+    account = accounts(:depository)
+
+    patch import_url(pdf_import), params: { import: { account_id: account.id } }
+
+    assert_redirected_to import_url(pdf_import)
+    assert_equal I18n.t("imports.update.account_saved", default: "Account saved."), flash[:notice]
+    assert_equal account, pdf_import.reload.account
+    assert_equal account, statement.reload.account
+  end
+
+  test "read only shared user cannot link source statement through pdf import account update" do
+    account = accounts(:credit_card)
+    statement = AccountStatement.create_from_upload!(
+      family: @user.family,
+      account: nil,
+      file: uploaded_file(
+        filename: "statement.pdf",
+        content_type: "application/pdf",
+        content: file_fixture("imports/sample_bank_statement.pdf").binread
+      )
+    )
+    pdf_import = PdfImport.create_from_statement!(statement: statement)
+
+    sign_in users(:family_member)
+    patch import_url(pdf_import), params: { import: { account_id: account.id } }
+
+    assert_redirected_to account_url(account)
+    assert_equal I18n.t("accounts.not_authorized"), flash[:alert]
+    assert_nil pdf_import.reload.account
+    assert_nil statement.reload.account
+  end
+
+  test "user cannot view statement backed pdf import for inaccessible statement" do
+    statement = AccountStatement.create_from_upload!(
+      family: @user.family,
+      account: accounts(:investment),
+      file: uploaded_file(
+        filename: "statement.pdf",
+        content_type: "application/pdf",
+        content: file_fixture("imports/sample_bank_statement.pdf").binread
+      )
+    )
+    pdf_import = PdfImport.create_from_statement!(statement: statement)
+
+    sign_in users(:family_member)
+    get import_url(pdf_import)
+
+    assert_response :not_found
+  end
+
+  test "read only shared user cannot publish statement backed pdf import" do
+    account = accounts(:credit_card)
+    statement = AccountStatement.create_from_upload!(
+      family: @user.family,
+      account: account,
+      file: uploaded_file(
+        filename: "statement.pdf",
+        content_type: "application/pdf",
+        content: file_fixture("imports/sample_bank_statement.pdf").binread
+      )
+    )
+    pdf_import = PdfImport.create_from_statement!(statement: statement)
+    PdfImport.any_instance.expects(:publish_later).never
+
+    sign_in users(:family_member)
+    post publish_import_url(pdf_import)
+
+    assert_redirected_to account_url(account)
+    assert_equal I18n.t("accounts.not_authorized"), flash[:alert]
+  end
+
   test "rejects unsupported document type for DocumentImport option" do
     adapter = mock("vector_store_adapter")
     adapter.stubs(:supported_extensions).returns(%w[.pdf .txt])
diff --git a/test/jobs/process_pdf_job_test.rb b/test/jobs/process_pdf_job_test.rb
index 29d4a4977..1d9540dcc 100644
--- a/test/jobs/process_pdf_job_test.rb
+++ b/test/jobs/process_pdf_job_test.rb
@@ -18,12 +18,22 @@ class ProcessPdfJobTest < ActiveJob::TestCase
 
   test "skips if PDF not uploaded" do
     assert_not @import.pdf_uploaded?
+    @import.update_columns(status: "importing", updated_at: 31.minutes.ago)
 
     ProcessPdfJob.perform_now(@import)
 
     assert_equal "pending", @import.reload.status
   end
 
+  test "skips if PDF not uploaded without releasing fresh processing claim" do
+    assert_not @import.pdf_uploaded?
+    @import.update!(status: :importing)
+
+    ProcessPdfJob.perform_now(@import)
+
+    assert_equal "importing", @import.reload.status
+  end
+
   test "skips if already processed" do
     processed_import = imports(:pdf_processed)
 
@@ -33,6 +43,29 @@ class ProcessPdfJobTest < ActiveJob::TestCase
     assert_equal "complete", processed_import.reload.status
   end
 
+  test "skips already processed importing import and releases processing claim" do
+    processed_import = imports(:pdf_processed)
+    attach_pdf!(processed_import)
+    processed_import.update!(document_type: "financial_document")
+    processed_import.update_columns(status: "importing", updated_at: 31.minutes.ago)
+    processed_import.expects(:process_with_ai).never
+
+    ProcessPdfJob.perform_now(processed_import)
+
+    assert_equal "pending", processed_import.reload.status
+  end
+
+  test "skips already processed importing import without releasing fresh processing claim" do
+    processed_import = imports(:pdf_processed)
+    attach_pdf!(processed_import)
+    processed_import.update!(status: :importing, document_type: "financial_document")
+    processed_import.expects(:process_with_ai).never
+
+    ProcessPdfJob.perform_now(processed_import)
+
+    assert_equal "importing", processed_import.reload.status
+  end
+
   test "uploads non-bank PDF to vector store with classified type metadata" do
     pdf_content = attach_pdf!(@import)
     process_result = Struct.new(:document_type).new("financial_document")
diff --git a/test/models/account_statement_test.rb b/test/models/account_statement_test.rb
index a2829a0c9..5a24a1dd9 100644
--- a/test/models/account_statement_test.rb
+++ b/test/models/account_statement_test.rb
@@ -6,6 +6,8 @@ class AccountStatementTest < ActiveSupport::TestCase
     @account = accounts(:depository)
   end
 
+  LedgerBalance = Struct.new(:start_balance, :end_balance, keyword_init: true)
+
   OversizedDeclaredUpload = Struct.new(:original_filename, keyword_init: true) do
     def size
       AccountStatement::MAX_FILE_SIZE + 1
@@ -261,21 +263,23 @@ def rewind
   end
 
   test "rejects unsupported file extension even when mime type is broadly allowed" do
-    assert_raises(AccountStatement::InvalidUploadError) do
+    error = assert_raises(AccountStatement::InvalidUploadError) do
       AccountStatement.create_from_upload!(
         family: @family,
         account: @account,
         file: uploaded_file(filename: "statement.txt", content_type: "text/plain", content: "date,amount\n2024-01-01,1\n")
       )
     end
+    assert_equal :unsupported_type, error.reason
 
-    assert_raises(AccountStatement::InvalidUploadError) do
+    error = assert_raises(AccountStatement::InvalidUploadError) do
       AccountStatement.create_from_upload!(
         family: @family,
         account: @account,
         file: uploaded_file(filename: "statement.xls", content_type: "application/vnd.ms-excel", content: "date,amount\n2024-01-01,1\n")
       )
     end
+    assert_equal :unsupported_type, error.reason
   end
 
   test "rejects empty csv and xlsx statement uploads" do
@@ -288,17 +292,19 @@ def rewind
       )
     ].each do |file|
       assert_no_difference "AccountStatement.count" do
-        assert_raises(AccountStatement::InvalidUploadError) do
+        error = assert_raises(AccountStatement::InvalidUploadError) do
           AccountStatement.create_from_upload!(family: @family, account: @account, file: file)
         end
+        assert_equal :empty, error.reason
       end
     end
   end
 
   test "rejects declared oversized upload before reading content" do
-    assert_raises(AccountStatement::InvalidUploadError) do
+    error = assert_raises(AccountStatement::InvalidUploadError) do
       AccountStatement.prepare_upload!(OversizedDeclaredUpload.new(original_filename: "oversized.csv"))
     end
+    assert_equal :oversize, error.reason
   end
 
   test "streams unknown-size uploads and rejects when content exceeds size limit" do
@@ -308,9 +314,22 @@ def rewind
       content: "x" * (AccountStatement::MAX_FILE_SIZE + 1)
     )
 
-    assert_raises(AccountStatement::InvalidUploadError) do
+    error = assert_raises(AccountStatement::InvalidUploadError) do
       AccountStatement.prepare_upload!(file)
     end
+    assert_equal :oversize, error.reason
+  end
+
+  test "rejects pdf uploads without pdf header" do
+    error = assert_raises(AccountStatement::InvalidUploadError) do
+      AccountStatement.create_from_upload!(
+        family: @family,
+        account: @account,
+        file: uploaded_file(filename: "statement.pdf", content_type: "application/pdf", content: "not a pdf")
+      )
+    end
+
+    assert_equal :invalid_pdf_header, error.reason
   end
 
   test "stores sanitized csv parser output without raw rows" do
@@ -347,6 +366,21 @@ def rewind
     assert_equal Date.new(2024, 1, 31), statement.period_end_on
   end
 
+  test "uses a filename hint only as institution metadata" do
+    statement = AccountStatement.create_from_upload!(
+      family: @family,
+      account: nil,
+      file: uploaded_file(
+        filename: "Chase_2024-01.pdf",
+        content_type: "application/pdf",
+        content: "%PDF-1.4 statement"
+      )
+    )
+
+    assert_equal "Chase", statement.institution_name_hint
+    assert_nil statement.account_name_hint
+  end
+
   test "ignores unreasonable filename dates" do
     statement = AccountStatement.create_from_upload!(
       family: @family,
@@ -488,6 +522,52 @@ def rewind
     assert_equal "unavailable", statement.reconciliation_status
   end
 
+  test "reconciliation checks match when statement and ledger amounts align" do
+    statement = reconciliation_statement(opening_balance: "100.00", closing_balance: "150.00")
+    checks = statement.reconciliation_checks(
+      balance_lookup: reconciliation_lookup(opening_start: "100.00", closing_end: "150.00")
+    )
+
+    assert_reconciliation_statuses checks, {
+      "opening_balance" => "matched",
+      "closing_balance" => "matched",
+      "period_movement" => "matched"
+    }
+  end
+
+  test "reconciliation checks mismatch when amounts exceed tolerance" do
+    statement = reconciliation_statement(opening_balance: "100.00", closing_balance: "150.00")
+    checks = statement.reconciliation_checks(
+      balance_lookup: reconciliation_lookup(opening_start: "99.98", closing_end: "150.02")
+    )
+
+    assert_reconciliation_statuses checks, {
+      "opening_balance" => "mismatched",
+      "closing_balance" => "mismatched",
+      "period_movement" => "mismatched"
+    }
+  end
+
+  test "reconciliation checks treat exact one cent differences as matched" do
+    positive_checks = reconciliation_statement(opening_balance: "100.00", closing_balance: "150.00").reconciliation_checks(
+      balance_lookup: reconciliation_lookup(opening_start: "99.99", closing_end: "149.99")
+    )
+    negative_checks = reconciliation_statement(opening_balance: "100.00", closing_balance: "150.00").reconciliation_checks(
+      balance_lookup: reconciliation_lookup(opening_start: "100.01", closing_end: "150.01")
+    )
+
+    assert_reconciliation_statuses positive_checks, {
+      "opening_balance" => "matched",
+      "closing_balance" => "matched",
+      "period_movement" => "matched"
+    }
+    assert_reconciliation_statuses negative_checks, {
+      "opening_balance" => "matched",
+      "closing_balance" => "matched",
+      "period_movement" => "matched"
+    }
+  end
+
   test "coverage requires account" do
     error = assert_raises(ArgumentError) do
       AccountStatement::Coverage.new(nil)
@@ -582,23 +662,25 @@ def rewind
     assert statement.unmatched?
   end
 
-  test "preserves explicit rejected review status" do
+  test "does not reject linked statements" do
     statement = AccountStatement.create_from_upload!(
       family: @family,
       account: @account,
       file: uploaded_file(filename: "statement.csv", content_type: "text/csv", content: "date,amount\n2024-01-01,1\n")
     )
 
-    statement.reject_match!
+    assert_raises(AccountStatement::LinkedStatementRejectionError) do
+      statement.reject_match!
+    end
 
-    assert statement.rejected?
+    assert statement.linked?
     assert_equal @account, statement.account
   end
 
   test "preserves rejected review status across unrelated saves" do
     statement = AccountStatement.create_from_upload!(
       family: @family,
-      account: @account,
+      account: nil,
       file: uploaded_file(filename: "statement.csv", content_type: "text/csv", content: "date,amount\n2024-01-01,1\n")
     )
     statement.reject_match!
@@ -606,13 +688,13 @@ def rewind
     statement.update!(period_start_on: Date.new(2024, 1, 1))
 
     assert statement.rejected?
-    assert_equal @account, statement.account
+    assert_nil statement.account
   end
 
   test "allows intentional review status changes away from rejected" do
     statement = AccountStatement.create_from_upload!(
       family: @family,
-      account: @account,
+      account: nil,
       file: uploaded_file(filename: "statement.csv", content_type: "text/csv", content: "date,amount\n2024-01-01,1\n")
     )
     statement.reject_match!
@@ -711,6 +793,27 @@ def rewind
     end
   end
 
+  test "coverage start ignores statements without complete periods" do
+    account = Account.create!(
+      family: @family,
+      owner: users(:family_admin),
+      name: "Partial Archive Checking",
+      balance: 0,
+      currency: "USD",
+      accountable: Depository.new
+    )
+
+    create_statement(account: account, month: Date.new(2018, 1, 1), content: "linked-partial").update!(period_end_on: nil)
+    create_statement(account: nil, suggested_account: account, month: Date.new(2017, 1, 1), content: "suggested-partial").update!(period_end_on: nil)
+
+    travel_to Date.new(2026, 5, 6) do
+      coverage = AccountStatement::Coverage.for_year(account, nil)
+
+      assert_equal Date.new(2025, 5, 1), coverage.expected_start_month
+      assert_equal [ 2026, 2025 ], coverage.available_years
+    end
+  end
+
   test "coverage marks covered duplicate ambiguous and mismatched months" do
     covered_month = 5.months.ago.to_date.beginning_of_month
     missing_month = 4.months.ago.to_date.beginning_of_month
@@ -749,6 +852,37 @@ def rewind
 
   private
 
+    def reconciliation_statement(opening_balance:, closing_balance:)
+      AccountStatement.new(
+        family: @family,
+        account: @account,
+        period_start_on: Date.new(2024, 1, 1),
+        period_end_on: Date.new(2024, 1, 31),
+        opening_balance: opening_balance,
+        closing_balance: closing_balance,
+        currency: "USD"
+      )
+    end
+
+    def reconciliation_lookup(opening_start:, closing_end:)
+      opening_balance = LedgerBalance.new(start_balance: opening_start.to_d, end_balance: opening_start.to_d)
+      closing_balance = LedgerBalance.new(start_balance: closing_end.to_d, end_balance: closing_end.to_d)
+
+      lambda do |date, _currency|
+        case date
+        when Date.new(2024, 1, 1)
+          opening_balance
+        when Date.new(2024, 1, 31)
+          closing_balance
+        end
+      end
+    end
+
+    def assert_reconciliation_statuses(checks, expected)
+      statuses = checks.index_by { |check| check[:key] }.transform_values { |check| check[:status] }
+      assert_equal expected, statuses
+    end
+
     def create_statement(account:, month:, content:, suggested_account: nil, closing_balance: nil)
       statement = AccountStatement.create_from_upload!(
         family: @family,
diff --git a/test/models/pdf_import_test.rb b/test/models/pdf_import_test.rb
index e6d89bfe2..4eb0958cf 100644
--- a/test/models/pdf_import_test.rb
+++ b/test/models/pdf_import_test.rb
@@ -13,6 +13,16 @@ class PdfImportTest < ActiveSupport::TestCase
     assert_not @import.pdf_uploaded?
   end
 
+  test "pdf_uploaded? returns true for statement backed import" do
+    statement = create_pdf_statement
+    import = PdfImport.create_from_statement!(statement: statement)
+
+    assert import.pdf_uploaded?
+    assert import.statement_backed?
+    assert_equal statement.original_file.download, import.pdf_file_content
+    assert_equal statement.filename, import.pdf_filename
+  end
+
   test "ai_processed? returns false when no summary present" do
     assert_not @import.ai_processed?
   end
@@ -78,8 +88,26 @@ class PdfImportTest < ActiveSupport::TestCase
 
   test "process_with_ai_later enqueues ProcessPdfJob" do
     assert_enqueued_with job: ProcessPdfJob, args: [ @import ] do
-      @import.process_with_ai_later
+      assert @import.process_with_ai_later
     end
+
+    assert_equal "importing", @import.reload.status
+  end
+
+  test "process_with_ai_later does not enqueue duplicate jobs while importing" do
+    assert_enqueued_jobs 1, only: ProcessPdfJob do
+      assert @import.process_with_ai_later
+      assert_not @import.reload.process_with_ai_later
+    end
+
+    assert_equal "importing", @import.reload.status
+  end
+
+  test "process_with_ai_later resets pending when enqueue fails" do
+    ProcessPdfJob.stubs(:perform_later).raises(StandardError, "queue offline")
+
+    assert_not @import.process_with_ai_later
+    assert_equal "pending", @import.reload.status
   end
 
   test "generate_rows_from_extracted_data creates import rows" do
@@ -163,4 +191,100 @@ class PdfImportTest < ActiveSupport::TestCase
 
     assert_not ActiveStorage::Attachment.exists?(attachment_id)
   end
+
+  test "destroying statement backed import keeps statement file" do
+    statement = create_pdf_statement
+    import = PdfImport.create_from_statement!(statement: statement)
+    attachment_id = statement.original_file.id
+
+    perform_enqueued_jobs do
+      import.destroy!
+    end
+
+    assert ActiveStorage::Attachment.exists?(attachment_id)
+  end
+
+  test "statement backed import memoizes pdf content" do
+    statement = create_pdf_statement
+    import = PdfImport.create_from_statement!(statement: statement)
+    statement.original_file.expects(:download).once.returns("%PDF-test")
+
+    assert_equal "%PDF-test", import.pdf_file_content
+    assert_equal "%PDF-test", import.pdf_file_content
+  end
+
+  test "statement backed import reuse requires current account and date format" do
+    statement = create_pdf_statement
+    stale_import = PdfImport.create_from_statement!(statement: statement)
+    formats = Family::DATE_FORMATS.map(&:last)
+    alternate_date_format = (formats - [ statement.family.date_format ]).first || "#{statement.family.date_format}-alternate"
+    stale_import.update!(account: nil, date_format: alternate_date_format)
+
+    fresh_import = PdfImport.create_from_statement!(statement: statement)
+
+    assert_not_equal stale_import, fresh_import
+    assert_equal statement.account, fresh_import.account
+    assert_equal statement.family.date_format, fresh_import.date_format
+  end
+
+  test "statement backed import reuses matching reusable import" do
+    statement = create_pdf_statement
+    existing_import = PdfImport.create_from_statement!(statement: statement)
+
+    assert_equal existing_import, PdfImport.create_from_statement!(statement: statement)
+  end
+
+  test "assigning account links statement backed import statement" do
+    statement = create_pdf_statement(account: nil)
+    import = PdfImport.create_from_statement!(statement: statement)
+    account = accounts(:depository)
+
+    import.assign_account!(account)
+
+    assert_equal account, import.reload.account
+    assert_equal account, statement.reload.account
+    assert statement.linked?
+  end
+
+  test "statement backed import requires pdf statement" do
+    csv_statement = AccountStatement.create_from_upload!(
+      family: @import.family,
+      account: nil,
+      file: uploaded_file(filename: "statement.csv", content_type: "text/csv")
+    )
+    import = PdfImport.new(family: @import.family, account_statement: csv_statement)
+
+    assert_not import.valid?
+    assert import.errors[:account_statement].present?
+  end
+
+  test "statement backed import requires statement from same family" do
+    statement = AccountStatement.create_from_upload!(
+      family: families(:empty),
+      account: nil,
+      file: uploaded_file(
+        filename: "other_family_statement.pdf",
+        content_type: "application/pdf",
+        content: file_fixture("imports/sample_bank_statement.pdf").binread
+      )
+    )
+    import = PdfImport.new(family: @import.family, account_statement: statement)
+
+    assert_not import.valid?
+    assert import.errors[:account_statement].present?
+  end
+
+  private
+
+    def create_pdf_statement(account: accounts(:depository))
+      AccountStatement.create_from_upload!(
+        family: @import.family,
+        account: account,
+        file: uploaded_file(
+          filename: "sample_bank_statement.pdf",
+          content_type: "application/pdf",
+          content: file_fixture("imports/sample_bank_statement.pdf").binread
+        )
+      )
+    end
 end
diff --git a/test/system/settings_test.rb b/test/system/settings_test.rb
index 1107c2aed..238975368 100644
--- a/test/system/settings_test.rb
+++ b/test/system/settings_test.rb
@@ -27,7 +27,7 @@ class SettingsTest < ApplicationSystemTestCase
     # Add admin settings if user is admin
     if @user.admin?
       merchants_index = @settings_links.index([ "Merchants", family_merchants_path ])
-      @settings_links.insert(merchants_index + 1, [ "Statement Vault", account_statements_path ])
+      @settings_links.insert(merchants_index + 1, [ I18n.t("settings.settings_nav.statement_vault_label"), account_statements_path ])
       @settings_links += [
         [ "AI Prompts", settings_ai_prompts_path ],
         [ "API Key", settings_api_key_path ]
@@ -42,7 +42,9 @@ class SettingsTest < ApplicationSystemTestCase
       assert_current_path accounts_path, ignore_query: true
 
       @settings_links.each do |name, path|
-        click_link name, match: :first
+        within settings_navigation do
+          click_link name
+        end
         assert_selector "h1", text: name
         assert_current_path path
       end
@@ -95,7 +97,7 @@ class SettingsTest < ApplicationSystemTestCase
       assert_no_selector "li", text: "AI Prompts"
       assert_no_selector "li", text: "API Key"
       assert_no_selector "li", text: "Bank sync"
-      assert_no_selector "li", text: "Statement Vault"
+      assert_no_selector "li", text: I18n.t("settings.settings_nav.statement_vault_label")
     end
   end
 
@@ -108,4 +110,8 @@ def open_settings_from_sidebar
         click_link "Settings", match: :first
       end
     end
+
+    def settings_navigation
+      find("[data-testid='settings-nav']", match: :first, visible: :visible)
+    end
 end
diff --git a/test/test_helper.rb b/test/test_helper.rb
index 461ebe3a6..c6e1f1f5e 100644
--- a/test/test_helper.rb
+++ b/test/test_helper.rb
@@ -78,6 +78,10 @@ class TestCase
     # Setup all fixtures in test/fixtures/*.yml for all tests in alphabetical order.
     fixtures :all
 
+    teardown do
+      cleanup_uploaded_file_tempfiles
+    end
+
     # Add more helper methods to be used by all tests here...
     def sign_in(user)
       post sessions_path, params: { email: user.email, password: user_password_test }
@@ -112,7 +116,25 @@ def uploaded_file(filename:, content_type:, content: "date,amount\n2024-01-01,1\
       tempfile.write(content)
       tempfile.rewind
 
-      Rack::Test::UploadedFile.new(tempfile.path, content_type, true, original_filename: filename)
+      Rack::Test::UploadedFile.new(tempfile.path, content_type, true, original_filename: filename).tap do |uploaded_file|
+        uploaded_file.instance_variable_set(:@tempfile, tempfile)
+        uploaded_file_tempfiles << tempfile
+      end
+    end
+
+    def uploaded_file_tempfiles
+      @uploaded_file_tempfiles ||= []
+    end
+
+    def cleanup_uploaded_file_tempfiles
+      uploaded_file_tempfiles.each do |tempfile|
+        tempfile.close unless tempfile.closed?
+        tempfile.unlink if tempfile.path && File.exist?(tempfile.path)
+      rescue Errno::ENOENT
+        # File was already consumed and removed.
+      end
+    ensure
+      @uploaded_file_tempfiles = []
     end
 
     def family_guest