diff --git a/app/app/controllers/api/invitations_controller.rb b/app/app/controllers/api/invitations_controller.rb
index 2db732e63..d1008e8f5 100644
--- a/app/app/controllers/api/invitations_controller.rb
+++ b/app/app/controllers/api/invitations_controller.rb
@@ -4,14 +4,14 @@ class Api::InvitationsController < ApplicationController
   before_action :authenticate
 
   def create
-    @cbv_flow_invitation = CbvInvitationService.new(event_logger).invite(cbv_flow_invitation_params, @current_user, delivery_method: nil)
+    @cbv_flow_invitation = CbvInvitationService.new(event_logger)
+      .invite(cbv_flow_invitation_params, @current_user, delivery_method: nil)
 
-    if @cbv_flow_invitation.errors.any?
-      return render json: @cbv_flow_invitation.errors, status: :unprocessable_entity
+    errors = @cbv_flow_invitation.errors
+    if errors.any?
+      return render json: errors_to_json(errors), status: :unprocessable_entity
     end
 
-    @cbv_client = CbvClient.create_from_invitation(@cbv_flow_invitation)
-
     render json: {
       url: @cbv_flow_invitation.to_url,
       expiration_date: @cbv_flow_invitation.expires_at,
@@ -21,26 +21,30 @@ def create
 
   # can these be inferred from the model?
   def cbv_flow_invitation_params
+    if (applicant_attributes = params.delete(:agency_partner_metadata))
+      params[:cbv_applicant_attributes] = applicant_attributes.merge(client_agency_id: params[:client_agency_id])
+    end
+    params[:email_address] = @current_user.email
+
     params.permit(
-      :first_name,
-      :middle_name,
       :language,
-      :last_name,
-      :client_id_number,
-      :case_number,
       :email_address,
-      :snap_application_date,
-      :agency_id_number,
-      :beacon_id,
+      :client_agency_id,
       :user_id,
-      :client_agency_id
+      cbv_applicant_attributes: [
+        :first_name,
+        :middle_name,
+        :last_name,
+        :client_agency_id,
+        :client_id_number,
+        :case_number,
+        :snap_application_date,
+        :agency_id_number,
+        :beacon_id
+      ]
     )
   end
 
-  def client_agency_id
-    cbv_flow_invitation_params[:client_agency_id]
-  end
-
   private
 
   def authenticate
@@ -48,4 +52,25 @@ def authenticate
       @current_user = User.find_by_access_token(token)
     end
   end
+
+  def errors_to_json(errors)
+    # Generates a Hash of attribute => error_message and translates the
+    # internal names of objects (cbv_applicant) to the external names
+    # (agency_partner_metadata)
+    errors.map do |error|
+      next if error.attribute == :cbv_applicant
+
+      error_message = error.message
+
+      case error
+      when ActiveModel::NestedError
+        prefix, attribute_name = error.attribute.to_s.split(".")
+        prefix = "agency_partner_metadata" if prefix == "cbv_applicant"
+
+        [ "#{prefix}.#{attribute_name}", error_message ]
+      else
+        [ error.attribute, error_message ]
+      end
+    end.compact.to_h
+  end
 end
diff --git a/app/app/controllers/api/pinwheel_controller.rb b/app/app/controllers/api/pinwheel_controller.rb
index 2b48af46c..4b093464a 100644
--- a/app/app/controllers/api/pinwheel_controller.rb
+++ b/app/app/controllers/api/pinwheel_controller.rb
@@ -48,6 +48,7 @@ def user_action
     base_attributes = {
       timestamp: Time.now.to_i,
       cbv_flow_id: @cbv_flow.id,
+      cbv_applicant_id: @cbv_flow.cbv_applicant_id,
       invitation_id: @cbv_flow.cbv_flow_invitation_id
     }
     event_name = user_action_params[:event_name]
@@ -87,6 +88,7 @@ def token_params
   def track_event
     event_logger.track("ApplicantBeganLinkingEmployer", request, {
       cbv_flow_id: @cbv_flow.id,
+      cbv_applicant_id: @cbv_flow.cbv_applicant_id,
       invitation_id: @cbv_flow.cbv_flow_invitation_id,
       response_type: token_params[:response_type]
     })
diff --git a/app/app/controllers/caseworker/cbv_flow_invitations_controller.rb b/app/app/controllers/caseworker/cbv_flow_invitations_controller.rb
index a9940f3d6..b4cb5b24e 100644
--- a/app/app/controllers/caseworker/cbv_flow_invitations_controller.rb
+++ b/app/app/controllers/caseworker/cbv_flow_invitations_controller.rb
@@ -14,23 +14,24 @@ def new
   end
 
   def create
-    invitation_params = base_params.merge(client_agency_specific_params)
     # handle errors from the mail service
     begin
       @cbv_flow_invitation = CbvInvitationService.new(event_logger).invite(
-        invitation_params,
+        invitation_params.deep_merge(client_agency_id: client_agency_id, cbv_applicant_attributes: { client_agency_id: client_agency_id }),
         current_user,
         delivery_method: :email
       )
     rescue => e
       Rails.logger.error("Error inviting applicant: #{e.message}")
       flash[:alert] = t(".invite_failed",
-                        email_address: cbv_flow_invitation_params[:email_address],
+                        email_address: invitation_params[:email_address],
                         error_message: e.message)
       return redirect_to caseworker_dashboard_path(client_agency_id: params[:client_agency_id])
     end
 
     if @cbv_flow_invitation.errors.any?
+      @cbv_flow_invitation.errors.delete(:cbv_applicant)
+
       error_count = @cbv_flow_invitation.errors.size
       error_header = "#{helpers.pluralize(error_count, 'error')} occurred"
 
@@ -44,11 +45,8 @@ def create
       return render :new, status: :unprocessable_entity
     end
 
-    # hydrate the cbv_client with the invitation if there are no cbv_flow_invitation errors
-    @cbv_client = CbvClient.create_from_invitation(@cbv_flow_invitation)
-
     flash[:slim_alert] = {
-      message: t(".invite_success", email_address: cbv_flow_invitation_params[:email_address]),
+      message: t(".invite_success", email_address: invitation_params[:email_address]),
       type: "success"
     }
     redirect_to caseworker_dashboard_path(client_agency_id: params[:client_agency_id])
@@ -69,40 +67,20 @@ def ensure_valid_params!
     end
   end
 
-  def base_params
-    cbv_flow_invitation_params.slice(
-      :first_name,
-      :middle_name,
-      :language,
-      :last_name,
-      :email_address,
-      :snap_application_date,
-    ).merge(client_agency_id: client_agency_id)
-  end
-
-  def client_agency_specific_params
-    case client_agency_id
-    when "ma"
-      cbv_flow_invitation_params.slice(:agency_id_number, :beacon_id)
-    when "nyc"
-      cbv_flow_invitation_params.slice(:client_id_number, :case_number)
-    else
-      {}
-    end
-  end
-
-  def cbv_flow_invitation_params
+  def invitation_params
     params.fetch(:cbv_flow_invitation, {}).permit(
-      :first_name,
-      :middle_name,
       :language,
-      :last_name,
-      :client_id_number,
-      :case_number,
       :email_address,
-      :snap_application_date,
-      :agency_id_number,
-      :beacon_id,
+      cbv_applicant_attributes: [
+        :first_name,
+        :middle_name,
+        :last_name,
+        :client_id_number,
+        :case_number,
+        :snap_application_date,
+        :agency_id_number,
+        :beacon_id
+      ]
     )
   end
 
diff --git a/app/app/controllers/cbv/base_controller.rb b/app/app/controllers/cbv/base_controller.rb
index 5bc415622..0eaf756c3 100644
--- a/app/app/controllers/cbv/base_controller.rb
+++ b/app/app/controllers/cbv/base_controller.rb
@@ -47,7 +47,7 @@ def ensure_cbv_flow_not_yet_complete
   def current_agency
     return unless @cbv_flow.present? && @cbv_flow.client_agency_id.present?
 
-    @current_site ||= agency_config[@cbv_flow.client_agency_id]
+    @current_agency ||= agency_config[@cbv_flow.client_agency_id]
   end
 
   def next_path
@@ -110,6 +110,7 @@ def track_timeout_event
   def track_expired_event(invitation)
     event_logger.track("ApplicantLinkExpired", request, {
       invitation_id: invitation.id,
+      cbv_applicant_id: invitation.cbv_applicant_id,
       timestamp: Time.now.to_i
     })
   rescue => ex
@@ -121,6 +122,7 @@ def track_invitation_clicked_event(invitation, cbv_flow)
       timestamp: Time.now.to_i,
       invitation_id: invitation.id,
       cbv_flow_id: cbv_flow.id,
+      cbv_applicant_id: cbv_flow.cbv_applicant_id,
       client_agency_id: cbv_flow.client_agency_id,
       seconds_since_invitation: (Time.now - invitation.created_at).to_i
     })
diff --git a/app/app/controllers/cbv/employer_searches_controller.rb b/app/app/controllers/cbv/employer_searches_controller.rb
index 6943b3006..2b03bee06 100644
--- a/app/app/controllers/cbv/employer_searches_controller.rb
+++ b/app/app/controllers/cbv/employer_searches_controller.rb
@@ -31,6 +31,7 @@ def search_params
   def track_clicked_popular_payroll_providers_event
     event_logger.track("ApplicantClickedPopularPayrollProviders", request, {
       timestamp: Time.now.to_i,
+      cbv_applicant_id: @cbv_flow.cbv_applicant_id,
       cbv_flow_id: @cbv_flow.id,
       invitation_id: @cbv_flow.cbv_flow_invitation_id
     })
@@ -41,6 +42,7 @@ def track_clicked_popular_payroll_providers_event
   def track_clicked_popular_app_employers_event
     event_logger.track("ApplicantClickedPopularAppEmployers", request, {
       timestamp: Time.now.to_i,
+      cbv_applicant_id: @cbv_flow.cbv_applicant_id,
       cbv_flow_id: @cbv_flow.id,
       invitation_id: @cbv_flow.cbv_flow_invitation_id
     })
@@ -53,6 +55,7 @@ def track_accessed_search_event
 
     event_logger.track("ApplicantAccessedSearchPage", request, {
       timestamp: Time.now.to_i,
+      cbv_applicant_id: @cbv_flow.cbv_applicant_id,
       cbv_flow_id: @cbv_flow.id,
       invitation_id: @cbv_flow.cbv_flow_invitation_id
     })
@@ -65,6 +68,7 @@ def track_applicant_searched_event
 
     event_logger.track("ApplicantSearchedForEmployer", request, {
       timestamp: Time.now.to_i,
+      cbv_applicant_id: @cbv_flow.cbv_applicant_id,
       cbv_flow_id: @cbv_flow.id,
       invitation_id: @cbv_flow.cbv_flow_invitation_id,
       num_results: @employers.length,
diff --git a/app/app/controllers/cbv/entries_controller.rb b/app/app/controllers/cbv/entries_controller.rb
index 3033377c0..13fe5ab4b 100644
--- a/app/app/controllers/cbv/entries_controller.rb
+++ b/app/app/controllers/cbv/entries_controller.rb
@@ -3,6 +3,7 @@ def show
     event_logger.track("ApplicantViewedAgreement", request, {
       timestamp: Time.now.to_i,
       client_agency_id: @cbv_flow.client_agency_id,
+      cbv_applicant_id: @cbv_flow.cbv_applicant_id,
       cbv_flow_id: @cbv_flow.id,
       invitation_id: @cbv_flow.cbv_flow_invitation_id
     })
@@ -13,6 +14,7 @@ def create
       event_logger.track("ApplicantAgreed", request, {
         timestamp: Time.now.to_i,
         client_agency_id: @cbv_flow.client_agency_id,
+        cbv_applicant_id: @cbv_flow.cbv_applicant_id,
         cbv_flow_id: @cbv_flow.id,
         invitation_id: @cbv_flow.cbv_flow_invitation_id
       })
diff --git a/app/app/controllers/cbv/missing_results_controller.rb b/app/app/controllers/cbv/missing_results_controller.rb
index fd4e801ec..2dc46be5b 100644
--- a/app/app/controllers/cbv/missing_results_controller.rb
+++ b/app/app/controllers/cbv/missing_results_controller.rb
@@ -8,6 +8,7 @@ def show
   def track_missing_results_event
     event_logger.track("ApplicantAccessedMissingResultsPage", request, {
       timestamp: Time.now.to_i,
+      cbv_applicant_id: @cbv_flow.cbv_applicant_id,
       cbv_flow_id: @cbv_flow.id,
       invitation_id: @cbv_flow.cbv_flow_invitation_id
     })
diff --git a/app/app/controllers/cbv/payment_details_controller.rb b/app/app/controllers/cbv/payment_details_controller.rb
index bda7526e5..2bef3750a 100644
--- a/app/app/controllers/cbv/payment_details_controller.rb
+++ b/app/app/controllers/cbv/payment_details_controller.rb
@@ -126,6 +126,7 @@ def track_viewed_event
     return if @pinwheel_account.nil?
 
     event_logger.track("ApplicantViewedPaymentDetails", request, {
+      cbv_applicant_id: @cbv_flow.cbv_applicant_id,
       cbv_flow_id: @cbv_flow.id,
       invitation_id: @cbv_flow.cbv_flow_invitation_id,
       pinwheel_account_id: @pinwheel_account.id,
@@ -142,6 +143,7 @@ def track_saved_event
     comment_data = @cbv_flow.additional_information[params[:user][:account_id]]
 
     event_logger.track("ApplicantSavedPaymentDetails", request, {
+      cbv_applicant_id: @cbv_flow.cbv_applicant_id,
       cbv_flow_id: @cbv_flow.id,
       invitation_id: @cbv_flow.cbv_flow_invitation_id,
       additional_information_length: comment_data ? comment_data["comment"].length : 0
diff --git a/app/app/controllers/cbv/successes_controller.rb b/app/app/controllers/cbv/successes_controller.rb
index 06d1be6f1..9769bc5cc 100644
--- a/app/app/controllers/cbv/successes_controller.rb
+++ b/app/app/controllers/cbv/successes_controller.rb
@@ -8,6 +8,7 @@ def show
   def track_accessed_success_event
     event_logger.track("ApplicantAccessedSuccessPage", request, {
       timestamp: Time.now.to_i,
+      cbv_applicant_id: @cbv_flow.cbv_applicant_id,
       cbv_flow_id: @cbv_flow.id,
       invitation_id: @cbv_flow.cbv_flow_invitation_id
     })
diff --git a/app/app/controllers/cbv/summaries_controller.rb b/app/app/controllers/cbv/summaries_controller.rb
index d90798446..0fb376727 100644
--- a/app/app/controllers/cbv/summaries_controller.rb
+++ b/app/app/controllers/cbv/summaries_controller.rb
@@ -24,6 +24,7 @@ def show
         event_logger.track("ApplicantDownloadedIncomePDF", request, {
           timestamp: Time.now.to_i,
           client_agency_id: @cbv_flow.client_agency_id,
+          cbv_applicant_id: @cbv_flow.cbv_applicant_id,
           cbv_flow_id: @cbv_flow.id,
           invitation_id: @cbv_flow.cbv_flow_invitation_id,
           locale: I18n.locale
@@ -101,7 +102,7 @@ def transmit_to_caseworker
       time_now = Time.now
       beginning_date = (Date.parse(@payments_beginning_at).strftime("%b") rescue @payments_beginning_at)
       ending_date = (Date.parse(@payments_ending_at).strftime("%b%Y") rescue @payments_ending_at)
-      @file_name = "IncomeReport_#{@cbv_flow.cbv_flow_invitation.agency_id_number}_" \
+      @file_name = "IncomeReport_#{@cbv_flow.cbv_applicant.agency_id_number}_" \
         "#{beginning_date}-#{ending_date}_" \
         "Conf#{@cbv_flow.confirmation_code}_" \
         "#{time_now.strftime('%Y%m%d%H%M%S')}"
@@ -166,16 +167,16 @@ def generate_csv
     pinwheel_account = PinwheelAccount.find_by(cbv_flow_id: @cbv_flow.id)
 
     data = {
-      client_id: @cbv_flow.cbv_flow_invitation.agency_id_number,
-      first_name: @cbv_flow.cbv_flow_invitation.first_name,
-      last_name: @cbv_flow.cbv_flow_invitation.last_name,
-      middle_name: @cbv_flow.cbv_flow_invitation.middle_name,
+      client_id: @cbv_flow.cbv_applicant.agency_id_number,
+      first_name: @cbv_flow.cbv_applicant.first_name,
+      last_name: @cbv_flow.cbv_applicant.last_name,
+      middle_name: @cbv_flow.cbv_applicant.middle_name,
       client_email_address: @cbv_flow.cbv_flow_invitation.email_address,
-      beacon_userid: @cbv_flow.cbv_flow_invitation.beacon_id,
-      app_date: @cbv_flow.cbv_flow_invitation.snap_application_date.strftime("%m/%d/%Y"),
+      beacon_userid: @cbv_flow.cbv_applicant.beacon_id,
+      app_date: @cbv_flow.cbv_applicant.snap_application_date.strftime("%m/%d/%Y"),
       report_date_created: pinwheel_account.created_at.strftime("%m/%d/%Y"),
-      report_date_start: @cbv_flow.cbv_flow_invitation.paystubs_query_begins_at.strftime("%m/%d/%Y"),
-      report_date_end: @cbv_flow.cbv_flow_invitation.snap_application_date.strftime("%m/%d/%Y"),
+      report_date_start: @cbv_flow.cbv_applicant.paystubs_query_begins_at.strftime("%m/%d/%Y"),
+      report_date_end: @cbv_flow.cbv_applicant.snap_application_date.strftime("%m/%d/%Y"),
       confirmation_code: @cbv_flow.confirmation_code,
       consent_timestamp: @cbv_flow.consented_to_authorized_use_at.strftime("%m/%d/%Y %H:%M:%S"),
       pdf_filename: "#{@file_name}.pdf",
@@ -191,6 +192,7 @@ def track_transmitted_event(cbv_flow, payments)
     event_logger.track("ApplicantSharedIncomeSummary", request, {
       timestamp: Time.now.to_i,
       client_agency_id: cbv_flow.client_agency_id,
+      cbv_applicant_id: cbv_flow.cbv_applicant_id,
       cbv_flow_id: cbv_flow.id,
       invitation_id: cbv_flow.cbv_flow_invitation_id,
       account_count: cbv_flow.pinwheel_accounts.count,
@@ -209,6 +211,7 @@ def track_accessed_income_summary_event(cbv_flow, payments)
       timestamp: Time.now.to_i,
       client_agency_id: cbv_flow.client_agency_id,
       cbv_flow_id: cbv_flow.id,
+      cbv_applicant_id: cbv_flow.cbv_applicant_id,
       invitation_id: cbv_flow.cbv_flow_invitation_id,
       account_count: cbv_flow.pinwheel_accounts.count,
       paystub_count: payments.count,
diff --git a/app/app/controllers/help_controller.rb b/app/app/controllers/help_controller.rb
index 4d0b8d936..435292058 100644
--- a/app/app/controllers/help_controller.rb
+++ b/app/app/controllers/help_controller.rb
@@ -12,16 +12,19 @@ def show
 
     cbv_flow = session[:cbv_flow_id] ? CbvFlow.find_by(id: session[:cbv_flow_id]) : nil
 
-    event_logger.track("ApplicantViewedHelpTopic", request, {
-      topic: @help_topic,
-      cbv_flow_id: session[:cbv_flow_id],
-      invitation_id: cbv_flow&.cbv_flow_invitation_id,
-      client_agency_id: current_agency&.id,
-      flow_started_seconds_ago: cbv_flow ? (Time.now - cbv_flow.created_at).to_i : nil,
-      language: I18n.locale
-    }.compact)
-  rescue => ex
-    Rails.logger.error "Unable to track event (ApplicantViewedHelpTopic): #{ex}"
+    begin
+      event_logger.track("ApplicantViewedHelpTopic", request, {
+        topic: @help_topic,
+        cbv_applicant_id: cbv_flow&.cbv_applicant_id,
+        cbv_flow_id: session[:cbv_flow_id],
+        invitation_id: cbv_flow&.cbv_flow_invitation_id,
+        client_agency_id: current_agency&.id,
+        flow_started_seconds_ago: cbv_flow ? (Time.now - cbv_flow.created_at).to_i : nil,
+        language: I18n.locale
+      })
+    rescue => ex
+      Rails.logger.error "Unable to track event (ApplicantViewedHelpTopic): #{ex}"
+    end
   end
 
   private
diff --git a/app/app/controllers/webhooks/pinwheel/events_controller.rb b/app/app/controllers/webhooks/pinwheel/events_controller.rb
index 0377c4bfa..6c05ebf2a 100644
--- a/app/app/controllers/webhooks/pinwheel/events_controller.rb
+++ b/app/app/controllers/webhooks/pinwheel/events_controller.rb
@@ -59,6 +59,7 @@ def authorize_webhook
 
   def track_account_synced_event(cbv_flow, pinwheel_account)
     event_logger.track("ApplicantFinishedPinwheelSync", request, {
+      cbv_applicant_id: cbv_flow.cbv_applicant_id,
       cbv_flow_id: cbv_flow.id,
       invitation_id: cbv_flow.cbv_flow_invitation_id,
       identity_success: pinwheel_account.job_succeeded?("identity"),
@@ -77,6 +78,7 @@ def track_account_synced_event(cbv_flow, pinwheel_account)
 
   def track_account_created_event(cbv_flow, platform_name)
     event_logger.track("ApplicantCreatedPinwheelAccount", request, {
+      cbv_applicant_id: cbv_flow.cbv_applicant_id,
       cbv_flow_id: cbv_flow.id,
       invitation_id: cbv_flow.cbv_flow_invitation_id,
       platform_name: platform_name
diff --git a/app/app/helpers/cbv/pinwheel_data_helper.rb b/app/app/helpers/cbv/pinwheel_data_helper.rb
index 50ac827d7..25a27b06b 100644
--- a/app/app/helpers/cbv/pinwheel_data_helper.rb
+++ b/app/app/helpers/cbv/pinwheel_data_helper.rb
@@ -2,9 +2,9 @@ module Cbv::PinwheelDataHelper
   include ViewHelper
 
   def set_payments(account_id = nil)
-    invitation = @cbv_flow.cbv_flow_invitation
-    to_pay_date = invitation.snap_application_date
-    from_pay_date = invitation.paystubs_query_begins_at
+    applicant = @cbv_flow.cbv_applicant
+    to_pay_date = applicant.snap_application_date
+    from_pay_date = applicant.paystubs_query_begins_at
     @payments =
       if account_id.nil?
         fetch_paystubs(from_pay_date, to_pay_date)
@@ -36,7 +36,7 @@ def set_identities
       next unless pinwheel_account.job_succeeded?("identity")
 
       pinwheel.fetch_identity(account_id: pinwheel_account.pinwheel_account_id)
-    end
+    end.compact
   end
 
   def hours_by_earning_category(earnings)
diff --git a/app/app/mailers/application_mailer.rb b/app/app/mailers/application_mailer.rb
index 82a2b8744..12d6bfd82 100644
--- a/app/app/mailers/application_mailer.rb
+++ b/app/app/mailers/application_mailer.rb
@@ -18,6 +18,7 @@ def track_delivery
 
       # Include a couple attributes that are passed in as params to subclasses,
       # to help with linking metadata without including any PII.
+      cbv_applicant_id: params[:cbv_flow]&.cbv_applicant_id || params[:cbv_flow_invitation]&.cbv_applicant_id,
       cbv_flow_id: params[:cbv_flow]&.id,
       invitation_id: params[:cbv_flow_invitation]&.id
     })
diff --git a/app/app/mailers/weekly_report_mailer.rb b/app/app/mailers/weekly_report_mailer.rb
index ca11c2daa..143ccf7e4 100644
--- a/app/app/mailers/weekly_report_mailer.rb
+++ b/app/app/mailers/weekly_report_mailer.rb
@@ -33,28 +33,29 @@ def weekly_report_data(current_agency, report_range)
     CbvFlowInvitation
       .where(client_agency_id: current_agency.id)
       .where(created_at: report_range)
-      .includes(:cbv_flows)
+      .includes(:cbv_flows, :cbv_applicant)
       .map do |invitation|
         cbv_flow = invitation.cbv_flows.find(&:complete?)
+        applicant = invitation.cbv_applicant
 
         base_fields = {
           invited_at: invitation.created_at,
           transmitted_at: cbv_flow&.transmitted_at,
           completed_at: cbv_flow&.consented_to_authorized_use_at,
-          snap_application_date: invitation.snap_application_date,
+          snap_application_date: applicant.snap_application_date,
           email_address: invitation.email_address
         }
 
         case current_agency.id
         when "nyc"
           base_fields.merge(
-            client_id_number: invitation.client_id_number,
-            case_number: invitation.case_number,
+            client_id_number: applicant.client_id_number,
+            case_number: applicant.case_number,
           )
         when "ma"
           base_fields.merge(
-            agency_id_number: invitation.agency_id_number,
-            beacon_id: invitation.beacon_id
+            agency_id_number: applicant.agency_id_number,
+            beacon_id: applicant.beacon_id
           )
         end
       end
diff --git a/app/app/models/cbv_applicant.rb b/app/app/models/cbv_applicant.rb
new file mode 100644
index 000000000..e7cb2ef91
--- /dev/null
+++ b/app/app/models/cbv_applicant.rb
@@ -0,0 +1,140 @@
+class CbvApplicant < ApplicationRecord
+  # Massachusetts: 7 digits
+  MA_AGENCY_ID_REGEX = /\A\d{7}\z/
+
+  # Massachusetts: 6 alphanumeric characters
+  MA_BEACON_ID_REGEX = /\A[a-zA-Z0-9]{6}\z/
+
+  # New York City: 11 digits followed by 1 uppercase letter
+  NYC_CASE_NUMBER_REGEX = /\A\d{11}[A-Z]\z/
+
+  # New York City: 2 uppercase letters, followed by 5 digits, followed by 1 uppercase letter
+  NYC_CLIENT_ID_REGEX = /\A[A-Z]{2}\d{5}[A-Z]\z/
+
+  PAYSTUB_REPORT_RANGE = 90.days
+
+  has_many :cbv_flows
+  has_many :cbv_flow_invitations
+
+  before_validation :parse_snap_application_date
+  before_validation :format_case_number
+
+  validates :first_name, presence: true
+  validates :last_name, presence: true
+  validates :client_agency_id, presence: true
+
+  # MA specific validations
+  with_options(if: :ma_site?) do
+    validates :agency_id_number, format: { with: MA_AGENCY_ID_REGEX, message: :invalid_format }
+    validates :beacon_id, format: { with: MA_BEACON_ID_REGEX, message: :invalid_format }
+    validate :ma_snap_application_date_not_more_than_1_year_ago
+    validate :ma_snap_application_date_not_in_future
+  end
+
+  # NYC specific validations
+  with_options(if: :nyc_site?) do
+    validates :case_number, format: { with: NYC_CASE_NUMBER_REGEX, message: :invalid_format }
+    validates :client_id_number, format: { with: NYC_CLIENT_ID_REGEX, message: :invalid_format }
+    validate :nyc_snap_application_date_not_more_than_30_days_ago
+    validate :nyc_snap_application_date_not_in_future
+  end
+
+  include Redactable
+  has_redactable_fields(
+    first_name: :string,
+    middle_name: :string,
+    last_name: :string,
+    client_id_number: :string,
+    case_number: :string,
+    agency_id_number: :string,
+    beacon_id: :string,
+    snap_application_date: :date
+  )
+
+  def self.create_from_invitation(cbv_flow_invitation)
+    client = create!(
+      client_agency_id: cbv_flow_invitation.client_agency_id,
+      case_number: cbv_flow_invitation.case_number,
+      first_name: cbv_flow_invitation.first_name,
+      middle_name: cbv_flow_invitation.middle_name,
+      last_name: cbv_flow_invitation.last_name,
+      agency_id_number: cbv_flow_invitation.agency_id_number,
+      client_id_number: cbv_flow_invitation.client_id_number,
+      snap_application_date: cbv_flow_invitation.snap_application_date,
+      beacon_id: cbv_flow_invitation.beacon_id
+    )
+    cbv_flow_invitation.update_column(:cbv_applicant_id, client.id)
+    client
+  end
+
+  def ma_site?
+    client_agency_id == "ma"
+  end
+
+  def nyc_site?
+    client_agency_id == "nyc"
+  end
+
+  def parse_snap_application_date
+    raw_snap_application_date = @attributes["snap_application_date"]&.value_before_type_cast
+    return if raw_snap_application_date.is_a?(Date)
+
+    if raw_snap_application_date.is_a?(ActiveSupport::TimeWithZone) || raw_snap_application_date.is_a?(Time)
+      self.snap_application_date = raw_snap_application_date.to_date
+      # handle ISO 8601 date format, e.g. "2021-01-01" which is Ruby's default when querying a date field
+    elsif raw_snap_application_date.is_a?(String) && raw_snap_application_date.match?(/^\d{4}-\d{2}-\d{2}$/)
+      self.snap_application_date = Date.parse(raw_snap_application_date)
+    else
+      begin
+        new_date_format = Date.strptime(raw_snap_application_date.to_s, "%m/%d/%Y")
+        self.snap_application_date = new_date_format
+      rescue Date::Error => e
+        case client_agency_id
+        when "ma"
+          error = :ma_invalid_date
+        when "nyc"
+          error = :nyc_invalid_date
+        else
+          error = :default_invalid_date
+        end
+        errors.add(:snap_application_date, error)
+      end
+    end
+  end
+
+  def ma_snap_application_date_not_in_future
+    if snap_application_date.present? && snap_application_date > Date.current
+      errors.add(:snap_application_date, :ma_invalid_date)
+    end
+  end
+
+  def ma_snap_application_date_not_more_than_1_year_ago
+    if snap_application_date.present? && snap_application_date < 1.year.ago.to_date
+      errors.add(:snap_application_date, :ma_invalid_date)
+    end
+  end
+
+  def nyc_snap_application_date_not_in_future
+    if snap_application_date.present? && snap_application_date > Date.current
+      errors.add(:snap_application_date, :nyc_invalid_date)
+    end
+  end
+
+  def nyc_snap_application_date_not_more_than_30_days_ago
+    if snap_application_date.present? && snap_application_date < 30.day.ago.to_date
+      errors.add(:snap_application_date, :nyc_invalid_date)
+    end
+  end
+
+  def format_case_number
+    return if case_number.blank?
+    case_number.upcase!
+    if case_number.length == 9
+      self.case_number = "000#{case_number}"
+    end
+  end
+
+  def paystubs_query_begins_at
+    PAYSTUB_REPORT_RANGE.before(snap_application_date)
+  end
+end
diff --git a/app/app/models/cbv_client.rb b/app/app/models/cbv_client.rb
deleted file mode 100644
index 34af5a1bf..000000000
--- a/app/app/models/cbv_client.rb
+++ /dev/null
@@ -1,19 +0,0 @@
-class CbvClient < ApplicationRecord
-  has_one :cbv_flow
-  has_one :cbv_flow_invitation
-
-  def self.create_from_invitation(cbv_flow_invitation)
-    client = create!(
-      case_number: cbv_flow_invitation.case_number,
-      first_name: cbv_flow_invitation.first_name,
-      middle_name: cbv_flow_invitation.middle_name,
-      last_name: cbv_flow_invitation.last_name,
-      agency_id_number: cbv_flow_invitation.agency_id_number,
-      client_id_number: cbv_flow_invitation.client_id_number,
-      snap_application_date: cbv_flow_invitation.snap_application_date,
-      beacon_id: cbv_flow_invitation.beacon_id
-    )
-    cbv_flow_invitation.update_column(:cbv_client_id, client.id)
-    client
-  end
-end
diff --git a/app/app/models/cbv_flow.rb b/app/app/models/cbv_flow.rb
index a3f5d06b5..54d76ffb6 100644
--- a/app/app/models/cbv_flow.rb
+++ b/app/app/models/cbv_flow.rb
@@ -1,9 +1,11 @@
 class CbvFlow < ApplicationRecord
   has_many :pinwheel_accounts, dependent: :destroy
   belongs_to :cbv_flow_invitation, optional: true
-  belongs_to :cbv_client, optional: true
+  belongs_to :cbv_applicant, optional: true
   validates :client_agency_id, inclusion: Rails.application.config.client_agencies.client_agency_ids
 
+  accepts_nested_attributes_for :cbv_applicant
+
   scope :incomplete, -> { where(confirmation_code: nil) }
 
   include Redactable
@@ -20,8 +22,8 @@ def complete?
   def self.create_from_invitation(cbv_flow_invitation)
     create(
       cbv_flow_invitation: cbv_flow_invitation,
-      case_number: cbv_flow_invitation.case_number,
-      client_agency_id: cbv_flow_invitation.client_agency_id
+      cbv_applicant: cbv_flow_invitation.cbv_applicant,
+      client_agency_id: cbv_flow_invitation.client_agency_id,
     )
   end
 
diff --git a/app/app/models/cbv_flow_invitation.rb b/app/app/models/cbv_flow_invitation.rb
index d73645325..ee2712b69 100644
--- a/app/app/models/cbv_flow_invitation.rb
+++ b/app/app/models/cbv_flow_invitation.rb
@@ -8,59 +8,30 @@ class CbvFlowInvitation < ApplicationRecord
   # to be of practical use here.
   EMAIL_REGEX = /\A([\w+\-].?)+@[a-z\d\-]+(\.[a-z\d\-]+)*\.[a-z\d\-]+\z/i
 
-  # Massachusetts: 7 digits
-  MA_AGENCY_ID_REGEX = /\A\d{7}\z/
-
-  # Massachusetts: 6 alphanumeric characters
-  MA_BEACON_ID_REGEX = /\A[a-zA-Z0-9]{6}\z/
-
-  # New York City: 11 digits followed by 1 uppercase letter
-  NYC_CASE_NUMBER_REGEX = /\A\d{11}[A-Z]\z/
-
-  # New York City: 2 uppercase letters, followed by 5 digits, followed by 1 uppercase letter
-  NYC_CLIENT_ID_REGEX = /\A[A-Z]{2}\d{5}[A-Z]\z/
-
-  # Invitation validity time zone
   INVITATION_VALIDITY_TIME_ZONE = "America/New_York"
 
-  # Paystub report range
-  PAYSTUB_REPORT_RANGE = 90.days
-
   VALID_LOCALES = Rails.application.config.i18n.available_locales.map(&:to_s).freeze
 
   belongs_to :user
-  belongs_to :cbv_client, optional: true
+  belongs_to :cbv_applicant, optional: true
   has_many :cbv_flows
 
   has_secure_token :auth_token, length: 36
 
-  before_validation :parse_snap_application_date
-  before_validation :format_case_number, if: :nyc_site?
+  accepts_nested_attributes_for :cbv_applicant
+
   before_validation :normalize_language
+  before_create :copy_fields_back_from_cbv_applicant
 
   validates :client_agency_id, inclusion: Rails.application.config.client_agencies.client_agency_ids
-  validates :first_name, presence: true
-  validates :last_name, presence: true
   validates :email_address, format: { with: EMAIL_REGEX, message: :invalid_format }
+  validates_associated :cbv_applicant
   validates :language, inclusion: {
     in: VALID_LOCALES,
     message: :invalid_format,
     case_sensitive: false
   }
 
-  # MA specific validations
-  validates :agency_id_number, format: { with: MA_AGENCY_ID_REGEX, message: :invalid_format }, if: :ma_site?
-  validates :beacon_id, format: { with: MA_BEACON_ID_REGEX, message: :invalid_format }, if: :ma_site?
-  validate :ma_snap_application_date_not_more_than_1_year_ago, if: :ma_site?
-  validate :ma_snap_application_date_not_in_future, if: :ma_site?
-
-
-  # NYC specific validations
-  validates :case_number, format: { with: NYC_CASE_NUMBER_REGEX, message: :invalid_format }, if: :nyc_site?
-  validates :client_id_number, format: { with: NYC_CLIENT_ID_REGEX, message: :invalid_format }, if: :nyc_site?
-  validate :nyc_snap_application_date_not_more_than_30_days_ago, if: :nyc_site?
-  validate :nyc_snap_application_date_not_in_future, if: :nyc_site?
-
   include Redactable
   has_redactable_fields(
     first_name: :string,
@@ -77,6 +48,23 @@ class CbvFlowInvitation < ApplicationRecord
 
   scope :unstarted, -> { left_outer_joins(:cbv_flows).where(cbv_flows: { id: nil }) }
 
+  # Temporarily copy fields back to the invitation so we can not violate the
+  # NOT NULL constraints until we delete these columns.
+  def copy_fields_back_from_cbv_applicant
+    return unless cbv_applicant.present?
+
+    assign_attributes(
+      first_name: cbv_applicant.first_name,
+      middle_name: cbv_applicant.middle_name,
+      last_name: cbv_applicant.last_name,
+      client_id_number: cbv_applicant.client_id_number,
+      case_number: cbv_applicant.case_number,
+      agency_id_number: cbv_applicant.agency_id_number,
+      beacon_id: cbv_applicant.beacon_id,
+      snap_application_date: cbv_applicant.snap_application_date
+    )
+  end
+
   # Invitations are valid until 11:59pm Eastern Time on the (e.g.) 14th day
   # after sending the invitation.
   def expires_at
@@ -98,79 +86,6 @@ def to_url
     Rails.application.routes.url_helpers.cbv_flow_entry_url(token: auth_token, locale: language)
   end
 
-  def paystubs_query_begins_at
-    PAYSTUB_REPORT_RANGE.before(snap_application_date)
-  end
-
-  private
-
-  def nyc_site?
-    client_agency_id == "nyc"
-  end
-
-  def ma_site?
-    client_agency_id == "ma"
-  end
-
-  def parse_snap_application_date
-    raw_snap_application_date = @attributes["snap_application_date"]&.value_before_type_cast
-    return if raw_snap_application_date.is_a?(Date)
-
-    if raw_snap_application_date.is_a?(ActiveSupport::TimeWithZone) || raw_snap_application_date.is_a?(Time)
-      self.snap_application_date = raw_snap_application_date.to_date
-      # handle ISO 8601 date format, e.g. "2021-01-01" which is Ruby's default when querying a date field
-    elsif raw_snap_application_date.is_a?(String) && raw_snap_application_date.match?(/^\d{4}-\d{2}-\d{2}$/)
-      self.snap_application_date = Date.parse(raw_snap_application_date)
-    else
-      begin
-        new_date_format = Date.strptime(raw_snap_application_date.to_s, "%m/%d/%Y")
-        self.snap_application_date = new_date_format
-      rescue Date::Error => e
-        case client_agency_id
-        when "ma"
-          error = :ma_invalid_date
-        when "nyc"
-          error = :nyc_invalid_date
-        else
-          error = :default_invalid_date
-        end
-        errors.add(:snap_application_date, error)
-      end
-    end
-  end
-
-  def ma_snap_application_date_not_in_future
-    if snap_application_date.present? && snap_application_date > Date.current
-      errors.add(:snap_application_date, :ma_invalid_date)
-    end
-  end
-
-  def ma_snap_application_date_not_more_than_1_year_ago
-    if snap_application_date.present? && snap_application_date < 1.year.ago.to_date
-      errors.add(:snap_application_date, :ma_invalid_date)
-    end
-  end
-
-  def nyc_snap_application_date_not_in_future
-    if snap_application_date.present? && snap_application_date > Date.current
-      errors.add(:snap_application_date, :nyc_invalid_date)
-    end
-  end
-
-  def nyc_snap_application_date_not_more_than_30_days_ago
-    if snap_application_date.present? && snap_application_date < 30.day.ago.to_date
-      errors.add(:snap_application_date, :nyc_invalid_date)
-    end
-  end
-
-  def format_case_number
-    return if case_number.blank?
-    case_number.upcase!
-    if case_number.length == 9
-      self.case_number = "000#{case_number}"
-    end
-  end
-
   def normalize_language
     self.language = language.to_s.downcase if language.present?
   end
diff --git a/app/app/services/cbv_invitation_service.rb b/app/app/services/cbv_invitation_service.rb
index 4fe7ef211..a85af745d 100644
--- a/app/app/services/cbv_invitation_service.rb
+++ b/app/app/services/cbv_invitation_service.rb
@@ -35,6 +35,7 @@ def track_event(cbv_flow_invitation, current_user)
       user_id: current_user.id,
       caseworker_email_address: current_user.email,
       client_agency_id: cbv_flow_invitation.client_agency_id,
+      cbv_applicant_id: cbv_flow_invitation.cbv_applicant_id,
       invitation_id: cbv_flow_invitation.id
     })
   end
diff --git a/app/app/services/data_retention_service.rb b/app/app/services/data_retention_service.rb
index 963707d87..9038393e4 100644
--- a/app/app/services/data_retention_service.rb
+++ b/app/app/services/data_retention_service.rb
@@ -18,7 +18,10 @@ def redact_invitations
       .unstarted
       .unredacted
       .find_each do |cbv_flow_invitation|
-        cbv_flow_invitation.redact! if Time.now.after?(cbv_flow_invitation.expires_at + REDACT_UNUSED_INVITATIONS_AFTER)
+        next unless Time.now.after?(cbv_flow_invitation.expires_at + REDACT_UNUSED_INVITATIONS_AFTER)
+
+        cbv_flow_invitation.redact!
+        cbv_flow_invitation.cbv_applicant&.redact!
       end
   end
 
@@ -36,6 +39,7 @@ def redact_incomplete_cbv_flows
 
           cbv_flow.redact!
           cbv_flow.cbv_flow_invitation.redact!
+          cbv_flow.cbv_applicant&.redact!
         else
           # Redact standalone CbvFlow records some period after their last
           # update.
@@ -47,6 +51,7 @@ def redact_incomplete_cbv_flows
           next unless Time.now.after?(flow_redact_at)
 
           cbv_flow.redact!
+          cbv_flow.cbv_applicant&.redact!
         end
       end
   end
@@ -59,14 +64,16 @@ def redact_complete_cbv_flows
       .find_each do |cbv_flow|
         cbv_flow.redact!
         cbv_flow.cbv_flow_invitation.redact! if cbv_flow.cbv_flow_invitation.present?
+        cbv_flow.cbv_applicant&.redact!
       end
   end
 
   # Use after conducting a user test or other time we want to manually redact a
   # specific person's data in the system.
   def self.manually_redact_by_case_number!(case_number)
-    invitation = CbvFlowInvitation.find_by!(case_number: case_number)
-    invitation.redact!
-    invitation.cbv_flows.map(&:redact!)
+    applicant = CbvApplicant.find_by!(case_number: case_number)
+    applicant.redact!
+    applicant.cbv_flow_invitations.map(&:redact!)
+    applicant.cbv_flows.map(&:redact!)
   end
 end
diff --git a/app/app/views/caseworker/cbv_flow_invitations/_ma.html.erb b/app/app/views/caseworker/cbv_flow_invitations/_ma.html.erb
index b2c2764c9..3f1cdd2e7 100644
--- a/app/app/views/caseworker/cbv_flow_invitations/_ma.html.erb
+++ b/app/app/views/caseworker/cbv_flow_invitations/_ma.html.erb
@@ -1,35 +1,38 @@
-<%= f.text_field :first_name, label: t(".invite.first_name") %>
+<% cbv_applicant = @cbv_flow_invitation.cbv_applicant || @cbv_flow_invitation.build_cbv_applicant %>
+<%= f.fields_for :cbv_applicant, cbv_applicant do |f2| %>
+  <%= f2.text_field :first_name, label: t(".invite.first_name") %>
 
-<%= f.text_field :middle_name, label: t(".invite.middle_name") %>
+  <%= f2.text_field :middle_name, label: t(".invite.middle_name") %>
 
-<%= f.text_field :last_name, label: t(".invite.last_name") %>
+  <%= f2.text_field :last_name, label: t(".invite.last_name") %>
 
-<%= f.text_field :agency_id_number, label: t(".invite.agency_id_number"), hint: "Format: 1234567" %>
+  <%= f2.text_field :agency_id_number, label: t(".invite.agency_id_number"), hint: "Format: 1234567" %>
 
-<%= f.date_picker :snap_application_date, label: t(".invite.todays_date") %>
+  <%= f2.date_picker :snap_application_date, label: t(".invite.todays_date") %>
 
-<%= f.email_field :email_address, label: t(".invite.email_address") %>
+  <%= f.email_field :email_address, label: t(".invite.email_address") %>
 
-<%= f.text_field :beacon_id, label: t(".invite.beacon_id"), hint: "Format: abc123" %>
+  <%= f2.text_field :beacon_id, label: t(".invite.beacon_id"), hint: "Format: abc123" %>
 
-<div class="usa-form-group <%= "usa-form-group--error" if f.field_error(:language).present? %>">
-  <legend class="usa-legend"><%= t(".invite.language_label") %></legend>
-  <%= f.field_error :language %>
-  <div class="usa-radio-group">
-    <% language_options.each_with_index do |(value, label), index| %>
-      <div class="usa-radio usa-radio--tile">
-        <input
-          class="usa-radio__input usa-radio__input--tile"
-          id="language-<%= value %>-<%= index %>"
-          type="radio"
-          name="<%= f.object_name %>[language]"
-          value="<%= value %>"
-          <%= 'checked="checked"' if f.object.language == value %>
-        >
-        <label class="usa-radio__label" for="language-<%= value %>-<%= index %>">
-          <%= label %>
-        </label>
-      </div>
-    <% end %>
+  <div class="usa-form-group <%= "usa-form-group--error" if f.field_error(:language).present? %>">
+    <legend class="usa-legend"><%= t(".invite.language_label") %></legend>
+    <%= f.field_error :language %>
+    <div class="usa-radio-group">
+      <% language_options.each_with_index do |(value, label), index| %>
+        <div class="usa-radio usa-radio--tile">
+          <input
+            class="usa-radio__input usa-radio__input--tile"
+            id="language-<%= value %>-<%= index %>"
+            type="radio"
+            name="<%= f.object_name %>[language]"
+            value="<%= value %>"
+            <%= 'checked="checked"' if f.object.language == value %>
+          >
+          <label class="usa-radio__label" for="language-<%= value %>-<%= index %>">
+            <%= label %>
+          </label>
+        </div>
+      <% end %>
+    </div>
   </div>
-</div>
+<% end %>
diff --git a/app/app/views/caseworker/cbv_flow_invitations/_nyc.html.erb b/app/app/views/caseworker/cbv_flow_invitations/_nyc.html.erb
index 021f40fb3..9f2b06f5a 100644
--- a/app/app/views/caseworker/cbv_flow_invitations/_nyc.html.erb
+++ b/app/app/views/caseworker/cbv_flow_invitations/_nyc.html.erb
@@ -1,35 +1,38 @@
-<%= f.text_field :first_name, label: t(".invite.first_name") %>
+<% cbv_applicant = @cbv_flow_invitation.cbv_applicant || @cbv_flow_invitation.build_cbv_applicant %>
+<%= f.fields_for :cbv_applicant, cbv_applicant do |f2| %>
+  <%= f2.text_field :first_name, label: t(".invite.first_name") %>
 
-<%= f.text_field :middle_name, label: t(".invite.middle_name") %>
+  <%= f2.text_field :middle_name, label: t(".invite.middle_name") %>
 
-<%= f.text_field :last_name, label: t(".invite.last_name") %>
+  <%= f2.text_field :last_name, label: t(".invite.last_name") %>
 
-<%= f.text_field :client_id_number, label: t(".invite.client_id_number"), hint: "Format: AB12345C" %>
+  <%= f2.text_field :client_id_number, label: t(".invite.client_id_number"), hint: "Format: AB12345C" %>
 
-<%= f.text_field :case_number, label: t(".invite.case_number"), hint: "Format: 00012345678A" %>
+  <%= f2.text_field :case_number, label: t(".invite.case_number"), hint: "Format: 00012345678A" %>
 
-<%= f.date_picker :snap_application_date, label: t(".invite.snap_application_date") %>
+  <%= f2.date_picker :snap_application_date, label: t(".invite.snap_application_date") %>
 
-<%= f.email_field :email_address, label: t(".invite.email_address") %>
+  <%= f.email_field :email_address, label: t(".invite.email_address") %>
 
-<div class="usa-form-group <%= "usa-form-group--error" if f.field_error(:language).present? %>">
-  <legend class="usa-legend"><%= t(".invite.language_label") %></legend>
-  <%= f.field_error :language %>
-  <div class="usa-radio-group">
-    <% language_options.each_with_index do |(value, label), index| %>
-      <div class="usa-radio usa-radio--tile">
-        <input
-          class="usa-radio__input usa-radio__input--tile"
-          id="language-<%= value %>-<%= index %>"
-          type="radio"
-          name="<%= f.object_name %>[language]"
-          value="<%= value %>"
-          <%= 'checked="checked"' if f.object.language == value %>
-        >
-        <label class="usa-radio__label" for="language-<%= value %>-<%= index %>">
-          <%= label %>
-        </label>
-      </div>
-    <% end %>
+  <div class="usa-form-group <%= "usa-form-group--error" if f.field_error(:language).present? %>">
+    <legend class="usa-legend"><%= t(".invite.language_label") %></legend>
+    <%= f.field_error :language %>
+    <div class="usa-radio-group">
+      <% language_options.each_with_index do |(value, label), index| %>
+        <div class="usa-radio usa-radio--tile">
+          <input
+            class="usa-radio__input usa-radio__input--tile"
+            id="language-<%= value %>-<%= index %>"
+            type="radio"
+            name="<%= f.object_name %>[language]"
+            value="<%= value %>"
+            <%= 'checked="checked"' if f.object.language == value %>
+          >
+          <label class="usa-radio__label" for="language-<%= value %>-<%= index %>">
+            <%= label %>
+          </label>
+        </div>
+      <% end %>
+    </div>
   </div>
-</div>
+<% end %>
diff --git a/app/app/views/caseworker/cbv_flow_invitations/_sandbox.html.erb b/app/app/views/caseworker/cbv_flow_invitations/_sandbox.html.erb
index cb3b59707..82d9e68ec 100644
--- a/app/app/views/caseworker/cbv_flow_invitations/_sandbox.html.erb
+++ b/app/app/views/caseworker/cbv_flow_invitations/_sandbox.html.erb
@@ -1,39 +1,42 @@
-<%= f.text_field :first_name, label: t(".invite.first_name") %>
-
-<%= f.text_field :middle_name, label: t(".invite.middle_name") %>
-
-<%= f.text_field :last_name, label: t(".invite.last_name") %>
-
-<%= f.text_field :case_number, label: t(".invite.case_number") %>
-
-<%= f.text_field :beacon_id, label: t(".invite.beacon_id") %>
-
-<%= f.text_field :agency_id_number, label: t(".invite.agency_id_number") %>
-
-<%= f.text_field :client_id_number, label: t(".invite.client_id_number") %>
-
-<%= f.date_picker :snap_application_date, label: t(".invite.snap_application_date") %>
-
-<%= f.email_field :email_address, label: t(".invite.email_address") %>
-
-<div class="usa-form-group <%= "usa-form-group--error" if f.field_error(:language).present? %>">
-  <legend class="usa-legend"><%= t(".invite.language_label") %></legend>
-  <%= f.field_error :language %>
-  <div class="usa-radio-group">
-    <% language_options.each_with_index do |(value, label), index| %>
-      <div class="usa-radio usa-radio--tile">
-        <input
-          class="usa-radio__input usa-radio__input--tile"
-          id="language-<%= value %>-<%= index %>"
-          type="radio"
-          name="<%= f.object_name %>[language]"
-          value="<%= value %>"
-          <%= 'checked="checked"' if f.object.language == value %>
-        >
-        <label class="usa-radio__label" for="language-<%= value %>-<%= index %>">
-          <%= label %>
-        </label>
-      </div>
-    <% end %>
+<% cbv_applicant = @cbv_flow_invitation.cbv_applicant || @cbv_flow_invitation.build_cbv_applicant %>
+<%= f.fields_for :cbv_applicant, cbv_applicant do |f2| %>
+  <%= f2.text_field :first_name, label: t(".invite.first_name") %>
+
+  <%= f2.text_field :middle_name, label: t(".invite.middle_name") %>
+
+  <%= f2.text_field :last_name, label: t(".invite.last_name") %>
+
+  <%= f2.text_field :case_number, label: t(".invite.case_number") %>
+
+  <%= f2.text_field :beacon_id, label: t(".invite.beacon_id") %>
+
+  <%= f2.text_field :agency_id_number, label: t(".invite.agency_id_number") %>
+
+  <%= f2.text_field :client_id_number, label: t(".invite.client_id_number") %>
+
+  <%= f2.date_picker :snap_application_date, label: t(".invite.snap_application_date") %>
+
+  <%= f.email_field :email_address, label: t(".invite.email_address") %>
+
+  <div class="usa-form-group <%= "usa-form-group--error" if f.field_error(:language).present? %>">
+    <legend class="usa-legend"><%= t(".invite.language_label") %></legend>
+    <%= f.field_error :language %>
+    <div class="usa-radio-group">
+      <% language_options.each_with_index do |(value, label), index| %>
+        <div class="usa-radio usa-radio--tile">
+          <input
+            class="usa-radio__input usa-radio__input--tile"
+            id="language-<%= value %>-<%= index %>"
+            type="radio"
+            name="<%= f.object_name %>[language]"
+            value="<%= value %>"
+            <%= 'checked="checked"' if f.object.language == value %>
+          >
+          <label class="usa-radio__label" for="language-<%= value %>-<%= index %>">
+            <%= label %>
+          </label>
+        </div>
+      <% end %>
+    </div>
   </div>
-</div>
+<% end %>
diff --git a/app/app/views/cbv/summaries/show.pdf.erb b/app/app/views/cbv/summaries/show.pdf.erb
index 063f6594e..081617ee2 100644
--- a/app/app/views/cbv/summaries/show.pdf.erb
+++ b/app/app/views/cbv/summaries/show.pdf.erb
@@ -36,21 +36,21 @@
     <h3 class="margin-0"><%= t(".pdf.shared.client_information") %></h3>
   <% end %>
   <% if current_agency?(:ma) && !is_caseworker %>
-    <%= table.with_row(t(".pdf.client.agency_id_number"), @cbv_flow.cbv_flow_invitation.agency_id_number) %>
+    <%= table.with_row(t(".pdf.client.agency_id_number"), @cbv_flow.cbv_applicant.agency_id_number) %>
   <% end %>
   <% if is_caseworker %>
-    <%= table.with_row(t(".pdf.caseworker.first_name"), @cbv_flow.cbv_flow_invitation.first_name) %>
-    <%= table.with_row(t(".pdf.caseworker.middle_name"), @cbv_flow.cbv_flow_invitation.middle_name) %>
-    <%= table.with_row(t(".pdf.caseworker.last_name"), @cbv_flow.cbv_flow_invitation.last_name) %>
+    <%= table.with_row(t(".pdf.caseworker.first_name"), @cbv_flow.cbv_applicant.first_name) %>
+    <%= table.with_row(t(".pdf.caseworker.middle_name"), @cbv_flow.cbv_applicant.middle_name) %>
+    <%= table.with_row(t(".pdf.caseworker.last_name"), @cbv_flow.cbv_applicant.last_name) %>
 
     <% if current_agency?(:nyc) %>
-      <%= table.with_row(t(".pdf.caseworker.client_id_number"), @cbv_flow.cbv_flow_invitation.client_id_number) %>
-      <%= table.with_row(t(".pdf.caseworker.case_number"), @cbv_flow.cbv_flow_invitation.case_number) %>
+      <%= table.with_row(t(".pdf.caseworker.client_id_number"), @cbv_flow.cbv_applicant.client_id_number) %>
+      <%= table.with_row(t(".pdf.caseworker.case_number"), @cbv_flow.cbv_applicant.case_number) %>
     <% end %>
 
     <% if current_agency?(:ma) %>
-      <%= table.with_row(t(".pdf.caseworker.client_email_address"), @cbv_flow.cbv_flow_invitation.email_address) %>
-      <%= table.with_row(t(".pdf.caseworker.snap_agency_id"), @cbv_flow.cbv_flow_invitation.agency_id_number) %>
+      <%= table.with_row(t(".pdf.caseworker.client_email_address"), @cbv_flow.cbv_applicant.email_address) %>
+      <%= table.with_row(t(".pdf.caseworker.snap_agency_id"), @cbv_flow.cbv_applicant.agency_id_number) %>
     <% end %>
   <% end %>
 <% end %>
@@ -62,13 +62,13 @@
   <% if @cbv_flow.confirmation_code.present? %>
     <%= table.with_row(t(".pdf.shared.confirmation_code"), @cbv_flow.confirmation_code) %>
   <% end %>
-  <%= table.with_row(agency_translation(".application_or_recertification_date"), format_parsed_date(@cbv_flow.cbv_flow_invitation.snap_application_date)) %>
+  <%= table.with_row(agency_translation(".application_or_recertification_date"), format_parsed_date(@cbv_flow.cbv_applicant.snap_application_date)) %>
   <%= table.with_row(t(".pdf.client.date_created"), format_parsed_date(@cbv_flow.consented_to_authorized_use_at)) %>
-  <%= table.with_row(t(".pdf.client.date_range"), "#{format_parsed_date(@cbv_flow.cbv_flow_invitation.paystubs_query_begins_at)} to #{format_parsed_date(@cbv_flow.cbv_flow_invitation.snap_application_date)}") %>
+  <%= table.with_row(t(".pdf.client.date_range"), "#{format_parsed_date(@cbv_flow.cbv_applicant.paystubs_query_begins_at)} to #{format_parsed_date(@cbv_flow.cbv_applicant.snap_application_date)}") %>
   <% if is_caseworker %>
     <%= table.with_row(t(".pdf.caseworker.agreement_consent_timestamp"), @cbv_flow.consented_to_authorized_use_at) %>
     <% if current_agency?(:ma) %>
-      <%= table.with_row(t(".pdf.caseworker.staff_beacon_id_wel_id"), @cbv_flow.cbv_flow_invitation.beacon_id) %>
+      <%= table.with_row(t(".pdf.caseworker.staff_beacon_id_wel_id"), @cbv_flow.cbv_applicant.beacon_id) %>
     <% end %>
   <% end %>
 <% end %>
diff --git a/app/app/views/layouts/application.html.erb b/app/app/views/layouts/application.html.erb
index e3f816cae..47fecc734 100644
--- a/app/app/views/layouts/application.html.erb
+++ b/app/app/views/layouts/application.html.erb
@@ -23,7 +23,7 @@
             <div class="usa-alert usa-alert--info">
               <div class="usa-alert__body">
                 <div class="usa-alert__heading">Information</div>
-                <p class="usa-alert__text"><%= flash[:notice] %></p>
+                <div class="usa-alert__text"><%= flash[:notice] %></div>
               </div>
             </div>
           <% end %>
@@ -32,7 +32,7 @@
             <div class="usa-alert usa-alert--<%= flash[:alert_type] || "error" %>">
               <div class="usa-alert__body">
                 <div class="usa-alert__heading"><%= flash[:alert_heading] || "Error" %></div>
-                <p class="usa-alert__text"><%= flash[:alert].html_safe %></p>
+                <div class="usa-alert__text"><%= flash[:alert].html_safe %></div>
               </div>
             </div>
           <% end %>
@@ -40,13 +40,13 @@
           <% if flash[:slim_alert] %>
             <div class="usa-alert usa-alert--<%= flash[:slim_alert]["type"] %> usa-alert--slim">
               <div class="usa-alert__body">
-                <p class="usa-alert__text">
+                <div class="usa-alert__text">
                   <% if flash[:slim_alert]["message_html"] %>
                     <%= flash[:slim_alert]["message_html"].html_safe %>
                   <% elsif flash[:slim_alert]["message"] %>
                     <%= flash[:slim_alert]["message"] %>
                   <% end %>
-                </p>
+                </div>
               </div>
             </div>
           <% end %>
diff --git a/app/config/i18n-tasks.yml b/app/config/i18n-tasks.yml
index 1998923ce..d333824e7 100644
--- a/app/config/i18n-tasks.yml
+++ b/app/config/i18n-tasks.yml
@@ -129,6 +129,7 @@ ignore_missing:
     - "caseworker.entries.*"
     # Caseworker emails that are only supported in English:
     - "activerecord.errors.models.cbv_flow_invitation.*"
+    - "activerecord.errors.models.cbv_applicant.*"
     - "caseworker_mailer.summary_email.*"
     # Caseworker PDF strings that are only supported in English:
     - "cbv.summaries.show.pdf.caseworker.*"
diff --git a/app/config/locales/en.yml b/app/config/locales/en.yml
index 94ff67308..7e2b5803d 100644
--- a/app/config/locales/en.yml
+++ b/app/config/locales/en.yml
@@ -3,7 +3,7 @@ en:
   activerecord:
     errors:
       models:
-        cbv_flow_invitation:
+        cbv_applicant:
           attributes:
             agency_id_number:
               blank: Enter a valid agency ID number.
@@ -14,19 +14,21 @@ en:
               invalid_format: Enter the case number in the ANGIE/sPOS format, i.e. 00012345678A.
             client_id_number:
               invalid_format: Enter the client's CIN in the correct format, XX00000X.
-            email_address:
-              blank: Enter the client's email address.
-              invalid_format: Enter an email address in the correct format, like name@example.com
             first_name:
               blank: Enter the client's first name.
-            language:
-              invalid_format: Language must be either English (en) or Spanish (es).
             last_name:
               blank: Enter the client's last name.
             snap_application_date:
               default_invalid_date: Enter today's date or the date you contacted the client.
               ma_invalid_date: Enter today's date or the date you contacted the client. This date must be today or in the past year.
               nyc_invalid_date: SNAP interview date must be today or in the past 30 days.
+        cbv_flow_invitation:
+          attributes:
+            email_address:
+              blank: Enter the client's email address.
+              invalid_format: Enter an email address in the correct format, like name@example.com
+            language:
+              invalid_format: Language must be either English (en) or Spanish (es).
   applicant_mailer:
     invitation_email:
       body_1:
diff --git a/app/db/migrate/20250205195508_rename_cbv_clients_to_cbv_applicants.rb b/app/db/migrate/20250205195508_rename_cbv_clients_to_cbv_applicants.rb
new file mode 100644
index 000000000..6bfca8448
--- /dev/null
+++ b/app/db/migrate/20250205195508_rename_cbv_clients_to_cbv_applicants.rb
@@ -0,0 +1,7 @@
+class RenameCbvClientsToCbvApplicants < ActiveRecord::Migration[7.1]
+  def change
+    rename_table :cbv_clients, :cbv_applicants
+    rename_column :cbv_flows, :cbv_client_id, :cbv_applicant_id
+    rename_column :cbv_flow_invitations, :cbv_client_id, :cbv_applicant_id
+  end
+end
diff --git a/app/db/migrate/20250206012936_add_site_id_to_cbv_applicant.rb b/app/db/migrate/20250206012936_add_site_id_to_cbv_applicant.rb
new file mode 100644
index 000000000..306e4b625
--- /dev/null
+++ b/app/db/migrate/20250206012936_add_site_id_to_cbv_applicant.rb
@@ -0,0 +1,5 @@
+class AddSiteIdToCbvApplicant < ActiveRecord::Migration[7.1]
+  def change
+    add_column :cbv_applicants, :client_agency_id, :string
+  end
+end
diff --git a/app/db/schema.rb b/app/db/schema.rb
index cfa92f5d9..6b9391ed4 100644
--- a/app/db/schema.rb
+++ b/app/db/schema.rb
@@ -10,7 +10,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema[7.1].define(version: 2025_02_05_215840) do
+ActiveRecord::Schema[7.1].define(version: 2025_02_06_012936) do
   # These are extensions that must be enabled in order to support this database
   enable_extension "pgcrypto"
   enable_extension "plpgsql"
@@ -23,7 +23,7 @@
     t.datetime "updated_at", null: false
   end
 
-  create_table "cbv_clients", force: :cascade do |t|
+  create_table "cbv_applicants", force: :cascade do |t|
     t.string "case_number"
     t.string "first_name", null: false
     t.string "middle_name"
@@ -35,6 +35,7 @@
     t.datetime "redacted_at"
     t.datetime "created_at", null: false
     t.datetime "updated_at", null: false
+    t.string "client_agency_id"
   end
 
   create_table "cbv_flow_invitations", force: :cascade do |t|
@@ -55,8 +56,8 @@
     t.bigint "user_id"
     t.string "language"
     t.datetime "invitation_reminder_sent_at"
-    t.bigint "cbv_client_id"
-    t.index ["cbv_client_id"], name: "index_cbv_flow_invitations_on_cbv_client_id"
+    t.bigint "cbv_applicant_id"
+    t.index ["cbv_applicant_id"], name: "index_cbv_flow_invitations_on_cbv_applicant_id"
     t.index ["user_id"], name: "index_cbv_flow_invitations_on_user_id"
   end
 
@@ -72,10 +73,10 @@
     t.string "client_agency_id"
     t.string "confirmation_code"
     t.datetime "transmitted_at"
-    t.datetime "consented_to_authorized_use_at"
     t.datetime "redacted_at"
-    t.bigint "cbv_client_id"
-    t.index ["cbv_client_id"], name: "index_cbv_flows_on_cbv_client_id"
+    t.bigint "cbv_applicant_id"
+    t.index ["cbv_applicant_id"], name: "index_cbv_flows_on_cbv_applicant_id"
+    t.datetime "consented_to_authorized_use_at"
     t.index ["cbv_flow_invitation_id"], name: "index_cbv_flows_on_cbv_flow_invitation_id"
   end
 
diff --git a/app/lib/tasks/backfills.rake b/app/lib/tasks/backfills.rake
index af6759df7..d70d4bfed 100644
--- a/app/lib/tasks/backfills.rake
+++ b/app/lib/tasks/backfills.rake
@@ -1,10 +1,10 @@
 namespace :backfills do
-  desc "Back-fill cbv_clients from existing cbv data"
-  task cbv_clients: :environment do
+  desc "Back-fill cbv_applicants from existing cbv data"
+  task cbv_applicants: :environment do
     CbvFlowInvitation.transaction do
       CbvFlowInvitation.find_each do |cbv_flow_invitation|
-        cbv_client = CbvClient.create_from_invitation(cbv_flow_invitation)
-        cbv_flow_invitation.cbv_flows.update_all(cbv_client_id: cbv_client.id)
+        cbv_applicant = CbvApplicant.create_from_invitation(cbv_flow_invitation)
+        cbv_flow_invitation.cbv_flows.update_all(cbv_applicant_id: cbv_applicant.id)
       end
     end
   end
diff --git a/app/spec/controllers/api/invitations_controller_spec.rb b/app/spec/controllers/api/invitations_controller_spec.rb
index 8a8c64af4..c8cec48d7 100644
--- a/app/spec/controllers/api/invitations_controller_spec.rb
+++ b/app/spec/controllers/api/invitations_controller_spec.rb
@@ -9,33 +9,40 @@
     end
 
     let(:valid_params) do
-      attributes_for(:cbv_flow_invitation,
-        client_agency_id: "ma",
-        beacon_id: "ABC123",
-        agency_id_number: "7890120"
-      )
+      attributes_for(:cbv_flow_invitation, :ma).tap do |params|
+        params[:agency_partner_metadata] = attributes_for(:cbv_applicant, :ma)
+      end
     end
 
     before do
       request.headers["Authorization"] = "Bearer #{api_access_token.access_token}"
     end
 
-    it "creates an invitation" do
-      post :create, params: valid_params
+    it "creates an invitation with an associated cbv_applicant" do
+      expect do
+        post :create, params: valid_params
+      end.to change(CbvFlowInvitation, :count).by(1)
+        .and change(CbvApplicant, :count).by(1)
+
       expect(response).to have_http_status(:created)
       expect(JSON.parse(response.body).keys).to include("url")
     end
 
     context "invalid params" do
       let(:invalid_params) do
-        valid_params.except(:first_name)
+        valid_params[:agency_partner_metadata].delete(:first_name)
+        valid_params.delete(:client_agency_id)
+        valid_params
       end
 
       it "returns unprocessable entity" do
         post :create, params: invalid_params
 
         expect(response).to have_http_status(:unprocessable_entity)
-        expect(JSON.parse(response.body).keys).to include("first_name")
+        parsed_response = JSON.parse(response.body)
+        expect(parsed_response.keys).not_to include("cbv_applicant")
+        expect(parsed_response.keys).to include("client_agency_id")
+        expect(parsed_response.keys).to include("agency_partner_metadata.first_name")
       end
     end
 
diff --git a/app/spec/controllers/caseworker/cbv_flow_invitations_controller/auth_spec.rb b/app/spec/controllers/caseworker/cbv_flow_invitations_controller/auth_spec.rb
index 511ba912b..b9dc0f0b8 100644
--- a/app/spec/controllers/caseworker/cbv_flow_invitations_controller/auth_spec.rb
+++ b/app/spec/controllers/caseworker/cbv_flow_invitations_controller/auth_spec.rb
@@ -5,7 +5,9 @@
   let(:ma_user) { create(:user, email: "test@test.com", client_agency_id: 'ma') }
   let(:ma_params) { { client_agency_id: "ma" } }
   let(:valid_params) do
-    attributes_for(:cbv_flow_invitation, :nyc)
+    attributes_for(:cbv_flow_invitation, :nyc).merge(
+      cbv_applicant_attributes: attributes_for(:cbv_applicant, :nyc)
+    )
   end
 
   describe "#new" do
diff --git a/app/spec/controllers/caseworker/cbv_flow_invitations_controller/ma_spec.rb b/app/spec/controllers/caseworker/cbv_flow_invitations_controller/ma_spec.rb
index 42380965b..47f200e9b 100644
--- a/app/spec/controllers/caseworker/cbv_flow_invitations_controller/ma_spec.rb
+++ b/app/spec/controllers/caseworker/cbv_flow_invitations_controller/ma_spec.rb
@@ -35,7 +35,9 @@
 
   describe "#create" do
     let(:cbv_flow_invitation_params) do
-      attributes_for(:cbv_flow_invitation, client_agency_id: "ma", beacon_id: "ABC123", agency_id_number: "7890120")
+      attributes_for(:cbv_flow_invitation, :ma).merge(
+        cbv_applicant_attributes: attributes_for(:cbv_applicant, :ma)
+      )
     end
 
     it "creates a CbvFlowInvitation record with the ma fields" do
@@ -43,25 +45,29 @@
         client_agency_id: ma_params[:client_agency_id],
         cbv_flow_invitation: cbv_flow_invitation_params
       }
+      expect(response).to have_http_status(:found)
 
       invitation = CbvFlowInvitation.all.last
       expect(invitation.first_name).to eq("Jane")
       expect(invitation.middle_name).to eq("Sue")
       expect(invitation.last_name).to eq("Doe")
-      expect(invitation.agency_id_number).to eq("7890120")
+      expect(invitation.agency_id_number).to eq(cbv_flow_invitation_params[:cbv_applicant_attributes][:agency_id_number])
       expect(invitation.email_address).to eq("test@example.com")
-      expect(invitation.snap_application_date).to eq(Time.zone.today)
-      expect(invitation.beacon_id).to eq("ABC123")
+      expect(invitation.snap_application_date).to eq(Date.yesterday)
+      expect(invitation.beacon_id).to eq(cbv_flow_invitation_params[:cbv_applicant_attributes][:beacon_id])
     end
 
     it "requires the ma fields" do
+      cbv_flow_invitation_params[:cbv_applicant_attributes].delete(:beacon_id)
+      cbv_flow_invitation_params[:cbv_applicant_attributes].delete(:agency_id_number)
+
       post :create, params: {
         client_agency_id: "ma",
-        cbv_flow_invitation: cbv_flow_invitation_params.except(:beacon_id, :agency_id_number)
+        cbv_flow_invitation: cbv_flow_invitation_params
       }
       expected_errors = [
-        I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.agency_id_number.invalid_format'),
-        I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.beacon_id.invalid_format')
+        I18n.t('activerecord.errors.models.cbv_applicant.attributes.agency_id_number.invalid_format'),
+        I18n.t('activerecord.errors.models.cbv_applicant.attributes.beacon_id.invalid_format')
       ]
       expected_error_message = "<ul><li>#{expected_errors.join('</li><li>')}</li></ul>"
       expect(flash[:alert]).to eq(expected_error_message)
diff --git a/app/spec/controllers/caseworker/cbv_flow_invitations_controller/nyc_spec.rb b/app/spec/controllers/caseworker/cbv_flow_invitations_controller/nyc_spec.rb
index d54c2d3f8..ba58a9ba3 100644
--- a/app/spec/controllers/caseworker/cbv_flow_invitations_controller/nyc_spec.rb
+++ b/app/spec/controllers/caseworker/cbv_flow_invitations_controller/nyc_spec.rb
@@ -30,7 +30,9 @@
 
   describe "#create" do
     let(:cbv_flow_invitation_params) do
-      attributes_for(:cbv_flow_invitation, :nyc)
+      attributes_for(:cbv_flow_invitation, :nyc).merge(
+        cbv_applicant_attributes: attributes_for(:cbv_applicant, :nyc)
+      )
     end
 
     it "creates a CbvFlowInvitation record with the nyc fields" do
@@ -43,17 +45,18 @@
       expect(invitation.first_name).to eq("Jane")
       expect(invitation.middle_name).to eq("Sue")
       expect(invitation.last_name).to eq("Doe")
-      expect(invitation.client_id_number).to eq(cbv_flow_invitation_params[:client_id_number])
-      expect(invitation.case_number).to eq(cbv_flow_invitation_params[:case_number])
+      expect(invitation.client_id_number).to eq(cbv_flow_invitation_params[:cbv_applicant_attributes][:client_id_number])
+      expect(invitation.case_number).to eq(cbv_flow_invitation_params[:cbv_applicant_attributes][:case_number])
       expect(invitation.email_address).to eq("test@example.com")
     end
 
     it "creates a CbvFlowInvitation record without optional fields" do
+      cbv_flow_invitation_params[:cbv_applicant_attributes].delete(:middle_name)
+
       post :create, params: {
         client_agency_id: nyc_params[:client_agency_id],
-        cbv_flow_invitation: cbv_flow_invitation_params.except(:middle_name)
+        cbv_flow_invitation: cbv_flow_invitation_params
       }
-      puts response.inspect
       invitation = CbvFlowInvitation.last
       expect(invitation.middle_name).to be_nil
     end
@@ -71,31 +74,31 @@
     # which does not play nice since there are multiple instances of the event logger.
 
     context "when validations succeed" do
-      it "creates a cbv_client record" do
+      it "creates a cbv_applicant record" do
         expect {
           post :create, params: {
             client_agency_id: nyc_params[:client_agency_id],
             cbv_flow_invitation: cbv_flow_invitation_params
           }
-        }.to change(CbvClient, :count).by(1)
+        }.to change(CbvApplicant, :count).by(1)
 
-        client = CbvClient.last
+        client = CbvApplicant.last
         expect(client.first_name).to eq("Jane")
       end
     end
 
     context "when validations fail" do
       before do
-        cbv_flow_invitation_params[:first_name] = nil
+        cbv_flow_invitation_params[:cbv_applicant_attributes][:first_name] = nil
       end
 
-      it "does not create a cbv_client record" do
+      it "does not create a cbv_applicant record" do
         expect {
           post :create, params: {
             client_agency_id: nyc_params[:client_agency_id],
             cbv_flow_invitation: cbv_flow_invitation_params
           }
-        }.to change(CbvClient, :count).by(0)
+        }.to change(CbvApplicant, :count).by(0)
       end
     end
   end
diff --git a/app/spec/controllers/cbv/employer_searches_controller_spec.rb b/app/spec/controllers/cbv/employer_searches_controller_spec.rb
index 02a88819f..21fd8dc86 100644
--- a/app/spec/controllers/cbv/employer_searches_controller_spec.rb
+++ b/app/spec/controllers/cbv/employer_searches_controller_spec.rb
@@ -25,6 +25,7 @@
           .to receive(:track)
           .with("ApplicantAccessedSearchPage", anything, hash_including(
             timestamp: be_a(Integer),
+            cbv_applicant_id: cbv_flow.cbv_applicant_id,
             cbv_flow_id: cbv_flow.id,
             invitation_id: cbv_flow.cbv_flow_invitation_id
           ))
@@ -36,6 +37,7 @@
           .to receive(:track)
           .with("ApplicantAccessedSearchPage", anything, hash_including(
             timestamp: be_a(Integer),
+            cbv_applicant_id: cbv_flow.cbv_applicant_id,
             cbv_flow_id: cbv_flow.id,
             invitation_id: cbv_flow.cbv_flow_invitation_id
           ))
@@ -47,6 +49,7 @@
           .to receive(:track)
           .with("ApplicantClickedPopularPayrollProviders", anything, hash_including(
             timestamp: be_a(Integer),
+            cbv_applicant_id: cbv_flow.cbv_applicant_id,
             cbv_flow_id: cbv_flow.id,
             invitation_id: cbv_flow.cbv_flow_invitation_id
           ))
@@ -58,6 +61,7 @@
           .to receive(:track)
           .with("ApplicantClickedPopularPayrollProviders", anything, hash_including(
             timestamp: be_a(Integer),
+            cbv_applicant_id: cbv_flow.cbv_applicant_id,
             cbv_flow_id: cbv_flow.id,
             invitation_id: cbv_flow.cbv_flow_invitation_id
           ))
@@ -69,6 +73,7 @@
           .to receive(:track)
           .with("ApplicantClickedPopularAppEmployers", anything, hash_including(
             timestamp: be_a(Integer),
+            cbv_applicant_id: cbv_flow.cbv_applicant_id,
             cbv_flow_id: cbv_flow.id,
             invitation_id: cbv_flow.cbv_flow_invitation_id
           ))
@@ -80,6 +85,7 @@
           .to receive(:track)
           .with("ApplicantClickedPopularAppEmployers", anything, hash_including(
             timestamp: be_a(Integer),
+            cbv_applicant_id: cbv_flow.cbv_applicant_id,
             cbv_flow_id: cbv_flow.id,
             invitation_id: cbv_flow.cbv_flow_invitation_id
           ))
@@ -130,6 +136,7 @@
         expect_any_instance_of(MixpanelEventTracker)
           .to receive(:track)
           .with("ApplicantSearchedForEmployer", anything, hash_including(
+            cbv_applicant_id: cbv_flow.cbv_applicant_id,
             cbv_flow_id: cbv_flow.id,
             invitation_id: cbv_flow.cbv_flow_invitation_id,
             num_results: 1,
@@ -142,6 +149,7 @@
         expect_any_instance_of(NewRelicEventTracker)
           .to receive(:track)
           .with("ApplicantSearchedForEmployer", anything, hash_including(
+            cbv_applicant_id: cbv_flow.cbv_applicant_id,
             cbv_flow_id: cbv_flow.id,
             invitation_id: cbv_flow.cbv_flow_invitation_id,
             num_results: 1,
diff --git a/app/spec/controllers/cbv/entries_controller_spec.rb b/app/spec/controllers/cbv/entries_controller_spec.rb
index 9ec7559f0..fb9476d3a 100644
--- a/app/spec/controllers/cbv/entries_controller_spec.rb
+++ b/app/spec/controllers/cbv/entries_controller_spec.rb
@@ -15,8 +15,10 @@
       let(:invitation) do
         create(
           :cbv_flow_invitation,
-          case_number: "ABC1234",
-          created_at: seconds_since_invitation.seconds.ago
+          created_at: seconds_since_invitation.seconds.ago,
+          cbv_applicant_attributes: {
+            case_number: "ABC1234"
+          }
         )
       end
 
@@ -38,7 +40,7 @@
 
         cbv_flow = invitation.cbv_flows.first
         expect(cbv_flow).to have_attributes(
-          case_number: "ABC1234",
+          cbv_applicant: have_attributes(case_number: "ABC1234"),
           cbv_flow_invitation: invitation
         )
       end
diff --git a/app/spec/controllers/cbv/summaries_controller_spec.rb b/app/spec/controllers/cbv/summaries_controller_spec.rb
index 8dab34507..320a10d73 100644
--- a/app/spec/controllers/cbv/summaries_controller_spec.rb
+++ b/app/spec/controllers/cbv/summaries_controller_spec.rb
@@ -7,8 +7,16 @@
   let(:supported_jobs) { %w[income paystubs employment identity] }
   let(:flow_started_seconds_ago) { 300 }
   let(:employment_errored_at) { nil }
-  let(:cbv_flow) { create(:cbv_flow, :with_pinwheel_account, created_at: flow_started_seconds_ago.seconds.ago, case_number: "ABC1234", supported_jobs: supported_jobs, employment_errored_at: employment_errored_at) }
-  let(:cbv_flow_invitation) { cbv_flow.cbv_flow_invitation }
+  let(:cbv_applicant) { create(:cbv_applicant) }
+  let(:cbv_flow) do
+    create(:cbv_flow,
+      :with_pinwheel_account,
+      created_at: flow_started_seconds_ago.seconds.ago,
+      supported_jobs: supported_jobs,
+      employment_errored_at: employment_errored_at,
+      cbv_applicant: cbv_applicant
+    )
+  end
   let(:mock_client_agency) { instance_double(ClientAgencyConfig::ClientAgency) }
   let(:nyc_user) { create(:user, email: "test@test.com", client_agency_id: 'nyc') }
   let(:ma_user) { create(:user, email: "test@example.com", client_agency_id: 'ma') }
@@ -22,7 +30,6 @@
       "public_key"        => @public_key
     })
 
-    session[:cbv_flow_invitation] = cbv_flow_invitation
     cbv_flow.pinwheel_accounts.first.update(pinwheel_account_id: "03e29160-f7e7-4a28-b2d8-813640e030d3")
   end
 
@@ -32,8 +39,7 @@
 
   describe "#show" do
     before do
-      cbv_flow_invitation.update(snap_application_date: Date.parse('2024-06-18'))
-      cbv_flow_invitation.update(created_at: Date.parse('2024-03-20'))
+      cbv_applicant.update(snap_application_date: Date.parse('2024-06-18'))
       session[:cbv_flow_id] = cbv_flow.id
       stub_request_end_user_accounts_response
       stub_request_end_user_paystubs_response
@@ -258,12 +264,12 @@
         end
 
         it "generates, gzips, encrypts, and uploads PDF and CSV files to S3" do
-          agency_id_number = cbv_flow_invitation.agency_id_number
-          beacon_id = cbv_flow_invitation.beacon_id
+          agency_id_number = cbv_applicant.agency_id_number
+          beacon_id = cbv_applicant.beacon_id
 
           expect(s3_service_double).to receive(:upload_file).once do |file_path, file_name|
             expect(file_path).to end_with('.gpg')
-            expect(file_name).to start_with("outfiles/IncomeReport_#{cbv_flow_invitation.agency_id_number}_")
+            expect(file_name).to start_with("outfiles/IncomeReport_#{cbv_applicant.agency_id_number}_")
             expect(file_name).to end_with('.tar.gz.gpg')
             expect(File.exist?(file_path)).to be true
           end
@@ -276,9 +282,9 @@
           end
 
           cbv_flow.update(client_agency_id: "ma")
-          cbv_flow_invitation.update(client_agency_id: "ma")
-          cbv_flow_invitation.update(beacon_id: beacon_id)
-          cbv_flow_invitation.update(agency_id_number: agency_id_number)
+          cbv_applicant.update(client_agency_id: "ma")
+          cbv_applicant.update(beacon_id: beacon_id)
+          cbv_applicant.update(agency_id_number: agency_id_number)
 
           patch :update
         end
diff --git a/app/spec/controllers/help_controller_spec.rb b/app/spec/controllers/help_controller_spec.rb
index d2fdfcbea..0ea2c5cb0 100644
--- a/app/spec/controllers/help_controller_spec.rb
+++ b/app/spec/controllers/help_controller_spec.rb
@@ -22,6 +22,7 @@
           .to receive(:track)
           .with("ApplicantViewedHelpTopic", be_an(ActionController::TestRequest), hash_including(
             browser: nil,
+            cbv_applicant_id: cbv_flow.cbv_applicant_id,
             cbv_flow_id: cbv_flow.id,
             device_name: nil,
             device_type: nil,
@@ -37,6 +38,7 @@
           .to receive(:track)
           .with("ApplicantViewedHelpTopic", be_an(ActionController::TestRequest), hash_including(
             browser: nil,
+            cbv_applicant_id: cbv_flow.cbv_applicant_id,
             cbv_flow_id: cbv_flow.id,
             device_name: nil,
             device_type: nil,
@@ -65,6 +67,7 @@
           .to receive(:track)
           .with("ApplicantViewedHelpTopic", be_an(ActionController::TestRequest), hash_including(
             browser: nil,
+            cbv_applicant_id: nil,
             cbv_flow_id: nil,
             device_name: nil,
             device_type: nil,
diff --git a/app/spec/factories/cbv_applicant.rb b/app/spec/factories/cbv_applicant.rb
new file mode 100644
index 000000000..e5a1739e9
--- /dev/null
+++ b/app/spec/factories/cbv_applicant.rb
@@ -0,0 +1,38 @@
+FactoryBot.define do
+  factory :cbv_applicant do
+    client_agency_id { "sandbox" }
+    first_name { "Jane" }
+    middle_name { "Sue" }
+    last_name { "Doe" }
+    snap_application_date { Date.yesterday.strftime("%m/%d/%Y") }
+
+    trait :nyc do
+      client_agency_id { "nyc" }
+
+      case_number do
+        number = 11.times.map { rand(10) }.join
+        letter = ('A'..'Z').to_a.sample
+        "#{number}#{letter}"
+      end
+
+      client_id_number do
+        letters = 2.times.map { ('A'..'Z').to_a.sample }.join
+        numbers = 5.times.map { rand(10) }.join
+        last_letter = ('A'..'Z').to_a.sample
+        "#{letters}#{numbers}#{last_letter}"
+      end
+    end
+
+    trait :ma do
+      client_agency_id { "ma" }
+
+      agency_id_number do
+        7.times.map { rand(10) }.join
+      end
+
+      beacon_id do
+        6.times.map { ('A'..'Z').to_a.sample }.join
+      end
+    end
+  end
+end
diff --git a/app/spec/factories/cbv_flow.rb b/app/spec/factories/cbv_flow.rb
index 31879601b..b6550ee26 100644
--- a/app/spec/factories/cbv_flow.rb
+++ b/app/spec/factories/cbv_flow.rb
@@ -1,6 +1,7 @@
 FactoryBot.define do
   factory :cbv_flow do
-    association :cbv_flow_invitation, factory: [ :cbv_flow_invitation ]
+    cbv_flow_invitation
+    cbv_applicant
 
     case_number { "ABC1234" }
     client_agency_id { "sandbox" }
@@ -22,5 +23,16 @@
         ]
       end
     end
+
+    transient do
+      cbv_applicant_attributes { {} }
+    end
+    after(:build) do |cbv_flow, evaluator|
+      cbv_flow.cbv_applicant.update(evaluator.cbv_applicant_attributes)
+
+      if cbv_flow.cbv_applicant && cbv_flow.cbv_flow_invitation
+        cbv_flow.cbv_flow_invitation.update(cbv_applicant: cbv_flow.cbv_applicant)
+      end
+    end
   end
 end
diff --git a/app/spec/factories/cbv_flow_invitation.rb b/app/spec/factories/cbv_flow_invitation.rb
index 29a44b14f..4c3ab3ee6 100644
--- a/app/spec/factories/cbv_flow_invitation.rb
+++ b/app/spec/factories/cbv_flow_invitation.rb
@@ -1,40 +1,31 @@
 FactoryBot.define do
   factory :cbv_flow_invitation, class: "CbvFlowInvitation" do
-    first_name { "Jane" }
-    middle_name { "Sue" }
-    language { :en }
-    last_name { "Doe" }
     client_agency_id { "sandbox" }
     email_address { "test@example.com" }
-    snap_application_date { Time.zone.today.strftime("%m/%d/%Y") }
+    language { :en }
     user
 
+    cbv_applicant
+
     trait :nyc do
       client_agency_id { "nyc" }
 
-      case_number do
-        number = 11.times.map { rand(10) }.join
-        letter = ('A'..'Z').to_a.sample
-        "#{number}#{letter}"
-      end
-
-      client_id_number do
-        letters = 2.times.map { ('A'..'Z').to_a.sample }.join
-        numbers = 5.times.map { rand(10) }.join
-        last_letter = ('A'..'Z').to_a.sample
-        "#{letters}#{numbers}#{last_letter}"
-      end
+      cbv_applicant { create(:cbv_applicant, :nyc) }
     end
 
     trait :ma do
       client_agency_id { "ma" }
 
-      agency_id_number do
-        7.times.map { rand(10) }.join
-      end
+      cbv_applicant { create(:cbv_applicant, :ma) }
+    end
+
+    transient do
+      cbv_applicant_attributes { {} }
+    end
 
-      beacon_id do
-        6.times.map { ('A'..'Z').to_a.sample }.join
+    after(:build) do |cbv_flow_invitation, evaluator|
+      if cbv_flow_invitation.cbv_applicant
+        cbv_flow_invitation.cbv_applicant.update(evaluator.cbv_applicant_attributes)
       end
     end
   end
diff --git a/app/spec/lib/backfills_spec.rb b/app/spec/lib/backfills_spec.rb
index 6a12c846f..9fbc22961 100644
--- a/app/spec/lib/backfills_spec.rb
+++ b/app/spec/lib/backfills_spec.rb
@@ -1,9 +1,9 @@
 require "rails_helper"
 
 RSpec.describe "backfills.rake" do
-  describe "backfills:cbv_clients" do
-    def expect_cbv_client_attributes_match(invitation)
-      expect(invitation.cbv_client).to have_attributes(
+  describe "backfills:cbv_applicants" do
+    def expect_cbv_applicant_attributes_match(invitation)
+      expect(invitation.cbv_applicant).to have_attributes(
         case_number: invitation.case_number,
         client_id_number: invitation.client_id_number,
         first_name: invitation.first_name,
@@ -23,33 +23,42 @@ def expect_cbv_client_attributes_match(invitation)
           language: nil,
           case_number: nil,
           client_id_number: nil,
-          snap_application_date: 31.days.ago.to_date
+          first_name: "Foo",
+          last_name: "Bar",
+          snap_application_date: 31.days.ago.to_date,
+          cbv_applicant: nil # intentionally don't create one to test the backfill
         })
         invitation.save(validate: false)
         invitation
       end
 
       let(:redacted_cbv_flow_invitation) do
-        invitation = create(:cbv_flow_invitation)
+        invitation = create(
+          :cbv_flow_invitation,
+          cbv_applicant: nil,
+          first_name: "Foo",
+          last_name: "Bar",
+          snap_application_date: "2024-01-01"
+        )
         invitation.redact!
         invitation
       end
 
-      it "Back-fills cbv_clients from an invalid cbv_flow_invitation" do
-        expect(invalid_cbv_flow_invitation.cbv_client).to be_nil
+      it "Back-fills cbv_applicants from an invalid cbv_flow_invitation" do
+        expect(invalid_cbv_flow_invitation.cbv_applicant).to be_nil
         expect(invalid_cbv_flow_invitation.valid?).to eq(false)
-        Rake::Task['backfills:cbv_clients'].execute
+        Rake::Task['backfills:cbv_applicants'].execute
         invalid_cbv_flow_invitation.reload
-        expect(invalid_cbv_flow_invitation.cbv_client).to be_present
-        expect_cbv_client_attributes_match(invalid_cbv_flow_invitation)
+        expect(invalid_cbv_flow_invitation.cbv_applicant).to be_present
+        expect_cbv_applicant_attributes_match(invalid_cbv_flow_invitation)
       end
 
-      it "Back-fills cbv_clients from a valid cbv_flow_invitation" do
-        expect(redacted_cbv_flow_invitation.cbv_client).to be_nil
-        Rake::Task['backfills:cbv_clients'].execute
+      it "Back-fills cbv_applicants from a valid cbv_flow_invitation" do
+        expect(redacted_cbv_flow_invitation.cbv_applicant).to be_nil
+        Rake::Task['backfills:cbv_applicants'].execute
         redacted_cbv_flow_invitation.reload
-        expect(redacted_cbv_flow_invitation.cbv_client).to be_present
-        expect_cbv_client_attributes_match(redacted_cbv_flow_invitation)
+        expect(redacted_cbv_flow_invitation.cbv_applicant).to be_present
+        expect_cbv_applicant_attributes_match(redacted_cbv_flow_invitation)
       end
     end
   end
diff --git a/app/spec/mailers/weekly_report_mailer_spec.rb b/app/spec/mailers/weekly_report_mailer_spec.rb
index 39d401ae5..26a869376 100644
--- a/app/spec/mailers/weekly_report_mailer_spec.rb
+++ b/app/spec/mailers/weekly_report_mailer_spec.rb
@@ -10,21 +10,22 @@
   let(:snap_app_date) { now.strftime("%Y-%m-%d") }
   let(:cbv_flow_invitation) do
     create(:cbv_flow_invitation,
-           client_agency_id.to_sym,
-           created_at: invitation_sent_at,
-           snap_application_date: invitation_sent_at - 1.day,
-          )
+      client_agency_id.to_sym,
+      created_at: invitation_sent_at
+    )
   end
   let(:cbv_flow) do
     create(
       :cbv_flow, :with_pinwheel_account,
-      case_number: cbv_flow_invitation.case_number,
       confirmation_code: "00001",
       created_at: invitation_sent_at + 15.minutes,
       client_agency_id: client_agency_id,
       transmitted_at: invitation_sent_at + 30.minutes,
-      cbv_flow_invitation_id: cbv_flow_invitation.id,
-      consented_to_authorized_use_at: invitation_sent_at + 30.minutes
+      cbv_flow_invitation: cbv_flow_invitation,
+      consented_to_authorized_use_at: invitation_sent_at + 30.minutes,
+      cbv_applicant_attributes: {
+        snap_application_date: invitation_sent_at - 1.day
+      }
     )
   end
   let(:mail) do
@@ -73,9 +74,9 @@
     expect(mail.attachments.first.content_type).to start_with('text/csv')
 
     expect(parsed_csv[0]).to match(
-      "client_id_number" => cbv_flow_invitation.client_id_number,
+      "client_id_number" => cbv_flow_invitation.cbv_applicant.client_id_number,
       "transmitted_at" => "2024-09-04 13:30:00 UTC",
-      "case_number" => cbv_flow.case_number,
+      "case_number" => cbv_flow_invitation.cbv_applicant.case_number,
       "invited_at" => "2024-09-04 13:00:00 UTC",
       "snap_application_date" => "2024-09-03",
       "completed_at" => "2024-09-04 13:30:00 UTC",
@@ -90,7 +91,7 @@
     it "excludes the record from the CSV" do
       expect(parsed_csv.length).to eq(0)
       expect(parsed_csv).not_to include(hash_including(
-        "client_id_number" => cbv_flow_invitation.client_id_number
+        "client_id_number" => cbv_flow_invitation.cbv_applicant.client_id_number
       ))
     end
   end
@@ -113,9 +114,9 @@
     it "includes them in the CSV data" do
       expect(parsed_csv.length).to eq(2)
       expect(parsed_csv).to include(hash_including(
-        "client_id_number" => incomplete_invitation.client_id_number,
+        "client_id_number" => incomplete_invitation.cbv_applicant.client_id_number,
         "transmitted_at" => nil,
-        "case_number" => incomplete_invitation.case_number,
+        "case_number" => incomplete_invitation.cbv_applicant.case_number,
         "invited_at" => "2024-09-04 13:00:00 UTC",
         "snap_application_date" => match(/\d\d\d\d-\d\d-\d\d/),
         "completed_at" => nil,
@@ -132,9 +133,9 @@
       expect(mail.attachments.first.content_type).to start_with('text/csv')
 
       expect(parsed_csv[0]).to match(
-        "beacon_id" => cbv_flow_invitation.beacon_id,
+        "beacon_id" => cbv_flow_invitation.cbv_applicant.beacon_id,
         "transmitted_at" => "2024-09-04 13:30:00 UTC",
-        "agency_id_number" => cbv_flow_invitation.agency_id_number,
+        "agency_id_number" => cbv_flow_invitation.cbv_applicant.agency_id_number,
         "invited_at" => "2024-09-04 13:00:00 UTC",
         "snap_application_date" => "2024-09-03",
         "completed_at" => "2024-09-04 13:30:00 UTC",
diff --git a/app/spec/models/cbv_applicant_spec.rb b/app/spec/models/cbv_applicant_spec.rb
new file mode 100644
index 000000000..f19062eb8
--- /dev/null
+++ b/app/spec/models/cbv_applicant_spec.rb
@@ -0,0 +1,193 @@
+require 'rails_helper'
+
+RSpec.describe CbvApplicant, type: :model do
+  describe '.create_from_invitation' do
+    let(:cbv_flow_invitation) { create(:cbv_flow_invitation) }
+
+    it 'creates a new CbvApplicant with attributes from cbv_flow_invitation' do
+      cbv_applicant = CbvApplicant.create_from_invitation(cbv_flow_invitation)
+      expect(cbv_applicant).to be_persisted
+      expect(cbv_applicant.case_number).to eq(cbv_flow_invitation.case_number)
+      expect(cbv_applicant.first_name).to eq(cbv_flow_invitation.first_name)
+      expect(cbv_applicant.middle_name).to eq(cbv_flow_invitation.middle_name)
+      expect(cbv_applicant.last_name).to eq(cbv_flow_invitation.last_name)
+      expect(cbv_applicant.agency_id_number).to eq(cbv_flow_invitation.agency_id_number)
+      expect(cbv_applicant.client_id_number).to eq(cbv_flow_invitation.client_id_number)
+      expect(cbv_applicant.snap_application_date).to eq(cbv_flow_invitation.snap_application_date)
+      expect(cbv_applicant.beacon_id).to eq(cbv_flow_invitation.beacon_id)
+    end
+
+    it 'associates the CbvApplicant with the CbvFlowInvitation' do
+      cbv_applicant = CbvApplicant.create_from_invitation(cbv_flow_invitation)
+      cbv_flow_invitation.reload
+      expect(cbv_flow_invitation.cbv_applicant).to eq(cbv_applicant)
+    end
+  end
+
+  describe "validations" do
+    let(:valid_attributes) { attributes_for(:cbv_applicant, :nyc) }
+
+    it "allows middle_name to be optional" do
+      applicant = create(:cbv_applicant, middle_name: nil)
+      expect(applicant).to be_valid
+    end
+
+    describe "snap_application_date" do
+      it "requires snap_application_date" do
+        applicant = CbvApplicant.new(valid_attributes.merge(snap_application_date: nil))
+        expect(applicant).not_to be_valid
+        expect(applicant.errors[:snap_application_date]).to include(
+          I18n.t('activerecord.errors.models.cbv_applicant.attributes.snap_application_date.nyc_invalid_date'),
+        )
+      end
+
+      it "validates snap_application_date is not in the future" do
+        applicant = CbvApplicant.new(valid_attributes.merge(snap_application_date: Date.tomorrow))
+        expect(applicant).not_to be_valid
+        expect(applicant.errors[:snap_application_date]).to include(
+          I18n.t('activerecord.errors.models.cbv_applicant.attributes.snap_application_date.nyc_invalid_date')
+        )
+      end
+
+      it "parses snap_application_date strings correctly" do
+        applicant = CbvApplicant.new(valid_attributes.merge(snap_application_date: "08/15/2023"))
+        expect(applicant).not_to be_valid
+        expect(applicant.snap_application_date).to eq(Date.new(2023, 8, 15))
+      end
+
+      it "adds an error when snap_application_date is not a valid date" do
+        applicant = CbvApplicant.new(valid_attributes.merge(snap_application_date: "invalid"))
+        expect(applicant).not_to be_valid
+        expect(applicant.errors[:snap_application_date]).to include(
+          I18n.t('activerecord.errors.models.cbv_applicant.attributes.snap_application_date.nyc_invalid_date')
+        )
+      end
+    end
+
+    context "when client_agency_id is 'nyc'" do
+      let(:nyc_attributes) { valid_attributes.merge(client_agency_id: 'nyc') }
+
+      context "user input is valid" do
+        it "formats a 9-character case number with leading zeros" do
+          applicant = CbvApplicant.new(nyc_attributes.merge(case_number: '12345678A'))
+          expect(applicant).to be_valid
+          expect(applicant.case_number).to eq('00012345678A')
+        end
+
+        it "converts case number to uppercase" do
+          applicant = CbvApplicant.new(nyc_attributes.merge(case_number: '12345678a'))
+          expect(applicant).to be_valid
+          expect(applicant.case_number).to eq('00012345678A')
+        end
+
+        it "validates snap_application_date is not older than 30 days" do
+          applicant = CbvApplicant.new(nyc_attributes.merge(snap_application_date: 31.days.ago))
+          expect(applicant).not_to be_valid
+          expect(applicant.errors[:snap_application_date]).to include(
+            I18n.t('activerecord.errors.models.cbv_applicant.attributes.snap_application_date.nyc_invalid_date')
+          )
+        end
+      end
+
+      context "user input is invalid" do
+        it "requires case_number" do
+          applicant = CbvApplicant.new(nyc_attributes.merge(case_number: nil))
+          expect(applicant).not_to be_valid
+          expect(applicant.errors[:case_number]).to include(
+            I18n.t('activerecord.errors.models.cbv_applicant.attributes.case_number.invalid_format'),
+          )
+        end
+
+        it "validates invalid case_number format" do
+          applicant = CbvApplicant.new(nyc_attributes.merge(case_number: 'invalid'))
+          expect(applicant).not_to be_valid
+          expect(applicant.errors[:case_number]).to include(
+            I18n.t('activerecord.errors.models.cbv_applicant.attributes.case_number.invalid_format')
+          )
+        end
+
+        it "checks that a shorter case number is invalid" do
+          applicant = CbvApplicant.new(nyc_attributes.merge(case_number: '123A'))
+          expect(applicant).not_to be_valid
+          expect(applicant.errors[:case_number]).to include(
+            I18n.t('activerecord.errors.models.cbv_applicant.attributes.case_number.invalid_format')
+          )
+        end
+
+        it "validates an invalid 11 char string" do
+          applicant = CbvApplicant.new(nyc_attributes.merge(case_number: '1234567890A'))
+          expect(applicant).not_to be_valid
+          expect(applicant.case_number).to eq('1234567890A')
+        end
+
+        it "validates client_id_number format when present" do
+          applicant = CbvApplicant.new(nyc_attributes.merge(client_id_number: 'invalid'))
+          expect(applicant).not_to be_valid
+          expect(applicant.errors[:client_id_number]).to include(
+            I18n.t('activerecord.errors.models.cbv_applicant.attributes.client_id_number.invalid_format')
+          )
+        end
+
+        it "requires valid snap_application_date" do
+          applicant = CbvApplicant.new(nyc_attributes.merge(snap_application_date: "invalid"))
+          expect(applicant).not_to be_valid
+          expect(applicant.errors[:snap_application_date]).to include(
+            I18n.t('activerecord.errors.models.cbv_applicant.attributes.snap_application_date.nyc_invalid_date')
+          )
+        end
+
+        it "requires client_id_number" do
+          applicant = CbvApplicant.new(nyc_attributes.merge(client_id_number: nil))
+          expect(applicant).not_to be_valid
+          expect(applicant.errors[:client_id_number]).to include(
+            I18n.t('activerecord.errors.models.cbv_applicant.attributes.client_id_number.invalid_format')
+          )
+        end
+      end
+    end
+
+    context "when client_agency_id is 'ma'" do
+      let(:ma_attributes) { valid_attributes.merge(client_agency_id: 'ma') }
+
+      context "user input is invalid" do
+        it "requires agency_id_number" do
+          applicant = CbvApplicant.new(ma_attributes)
+          expect(applicant).not_to be_valid
+          expect(applicant.errors[:agency_id_number]).to include(
+            I18n.t('activerecord.errors.models.cbv_applicant.attributes.agency_id_number.invalid_format'),
+          )
+        end
+
+        it "requires beacon_id" do
+          applicant = CbvApplicant.new(ma_attributes)
+          expect(applicant).not_to be_valid
+          expect(applicant.errors[:beacon_id]).to include(
+            I18n.t('activerecord.errors.models.cbv_applicant.attributes.beacon_id.invalid_format')
+          )
+        end
+
+        it "requires beacon_id to have 6 alphanumeric characters" do
+          applicant = CbvApplicant.new(ma_attributes.merge(beacon_id: '12345'))
+          expect(applicant).not_to be_valid
+          expect(applicant.errors[:beacon_id]).to include(
+            I18n.t('activerecord.errors.models.cbv_applicant.attributes.beacon_id.invalid_format')
+          )
+        end
+
+        it "validates agency_id_number format" do
+          applicant = CbvApplicant.new(ma_attributes.merge(agency_id_number: 'invalid'))
+          expect(applicant).not_to be_valid
+          expect(applicant.errors[:agency_id_number]).to include(
+            I18n.t('activerecord.errors.models.cbv_applicant.attributes.agency_id_number.invalid_format')
+          )
+        end
+
+        it "does not require client_id_number" do
+          applicant = CbvApplicant.new(valid_attributes.merge(client_id_number: nil, client_agency_id: "ma"))
+          expect(applicant).not_to be_valid
+          expect(applicant.errors[:client_id_number]).to be_empty
+        end
+      end
+    end
+  end
+end
diff --git a/app/spec/models/cbv_client_spec.rb b/app/spec/models/cbv_client_spec.rb
deleted file mode 100644
index 93f876475..000000000
--- a/app/spec/models/cbv_client_spec.rb
+++ /dev/null
@@ -1,26 +0,0 @@
-require 'rails_helper'
-
-RSpec.describe CbvClient, type: :model do
-  describe '.create_from_invitation' do
-    let(:cbv_flow_invitation) { create(:cbv_flow_invitation) }
-
-    it 'creates a new CbvClient with attributes from cbv_flow_invitation' do
-      cbv_client = CbvClient.create_from_invitation(cbv_flow_invitation)
-      expect(cbv_client).to be_persisted
-      expect(cbv_client.case_number).to eq(cbv_flow_invitation.case_number)
-      expect(cbv_client.first_name).to eq(cbv_flow_invitation.first_name)
-      expect(cbv_client.middle_name).to eq(cbv_flow_invitation.middle_name)
-      expect(cbv_client.last_name).to eq(cbv_flow_invitation.last_name)
-      expect(cbv_client.agency_id_number).to eq(cbv_flow_invitation.agency_id_number)
-      expect(cbv_client.client_id_number).to eq(cbv_flow_invitation.client_id_number)
-      expect(cbv_client.snap_application_date).to eq(cbv_flow_invitation.snap_application_date)
-      expect(cbv_client.beacon_id).to eq(cbv_flow_invitation.beacon_id)
-    end
-
-    it 'associates the CbvClient with the CbvFlowInvitation' do
-      cbv_client = CbvClient.create_from_invitation(cbv_flow_invitation)
-      cbv_flow_invitation.reload
-      expect(cbv_flow_invitation.cbv_client).to eq(cbv_client)
-    end
-  end
-end
diff --git a/app/spec/models/cbv_flow_invitation_spec.rb b/app/spec/models/cbv_flow_invitation_spec.rb
index 7be5f1744..2ba2ff42e 100644
--- a/app/spec/models/cbv_flow_invitation_spec.rb
+++ b/app/spec/models/cbv_flow_invitation_spec.rb
@@ -49,167 +49,6 @@
           I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.email_address.invalid_format')
         )
       end
-
-      it "requires snap_application_date" do
-        invitation = CbvFlowInvitation.new(valid_attributes.merge(snap_application_date: nil))
-        expect(invitation).not_to be_valid
-        expect(invitation.errors[:snap_application_date]).to include(
-          I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.snap_application_date.nyc_invalid_date'),
-        )
-      end
-
-      it "validates snap_application_date is not in the future" do
-        invitation = CbvFlowInvitation.new(valid_attributes.merge(snap_application_date: Date.tomorrow))
-        expect(invitation).not_to be_valid
-        expect(invitation.errors[:snap_application_date]).to include(
-          I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.snap_application_date.nyc_invalid_date')
-        )
-      end
-
-      it "parses snap_application_date strings correctly" do
-        invitation = CbvFlowInvitation.new(valid_attributes.merge(snap_application_date: "08/15/2023"))
-        expect(invitation).not_to be_valid
-        expect(invitation.snap_application_date).to eq(Date.new(2023, 8, 15))
-      end
-
-      it "adds an error when snap_application_date is not a valid date" do
-        invitation = CbvFlowInvitation.new(valid_attributes.merge(snap_application_date: "invalid"))
-        expect(invitation).not_to be_valid
-        expect(invitation.errors[:snap_application_date]).to include(
-          I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.snap_application_date.nyc_invalid_date')
-        )
-      end
-
-      it "allows middle_name to be optional" do
-        invitation = create(:cbv_flow_invitation, middle_name: nil)
-        expect(invitation).to be_valid
-      end
-    end
-
-    context "when client_agency_id is 'nyc'" do
-      let(:nyc_attributes) { valid_attributes.merge(client_agency_id: 'nyc', user: create(:user, client_agency_id: "nyc")) }
-
-      context "user input is valid" do
-        it "formats a 9-character case number with leading zeros" do
-          invitation = CbvFlowInvitation.new(nyc_attributes.merge(case_number: '12345678A'))
-          expect(invitation).to be_valid
-          expect(invitation.case_number).to eq('00012345678A')
-        end
-
-        it "converts case number to uppercase" do
-          invitation = CbvFlowInvitation.new(nyc_attributes.merge(case_number: '12345678a'))
-          expect(invitation).to be_valid
-          expect(invitation.case_number).to eq('00012345678A')
-        end
-
-        it "validates snap_application_date is not older than 30 days" do
-          invitation = CbvFlowInvitation.new(nyc_attributes.merge(snap_application_date: 31.days.ago))
-          expect(invitation).not_to be_valid
-          expect(invitation.errors[:snap_application_date]).to include(
-            I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.snap_application_date.nyc_invalid_date')
-          )
-        end
-      end
-
-      context "user input is invalid" do
-        it "requires case_number" do
-          invitation = CbvFlowInvitation.new(nyc_attributes.merge(case_number: nil))
-          expect(invitation).not_to be_valid
-          expect(invitation.errors[:case_number]).to include(
-            I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.case_number.invalid_format'),
-          )
-        end
-
-        it "validates invalid case_number format" do
-          invitation = CbvFlowInvitation.new(nyc_attributes.merge(case_number: 'invalid'))
-          expect(invitation).not_to be_valid
-          expect(invitation.errors[:case_number]).to include(
-            I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.case_number.invalid_format')
-          )
-        end
-
-        it "checks that a shorter case number is invalid" do
-          invitation = CbvFlowInvitation.new(nyc_attributes.merge(case_number: '123A'))
-          expect(invitation).not_to be_valid
-          expect(invitation.errors[:case_number]).to include(
-            I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.case_number.invalid_format')
-          )
-        end
-
-        it "validates an invalid 11 char string" do
-          invitation = CbvFlowInvitation.new(nyc_attributes.merge(case_number: '1234567890A'))
-          expect(invitation).not_to be_valid
-          expect(invitation.case_number).to eq('1234567890A')
-        end
-
-        it "validates client_id_number format when present" do
-          invitation = CbvFlowInvitation.new(nyc_attributes.merge(client_id_number: 'invalid'))
-          expect(invitation).not_to be_valid
-          expect(invitation.errors[:client_id_number]).to include(
-            I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.client_id_number.invalid_format')
-          )
-        end
-
-        it "requires valid snap_application_date" do
-          invitation = CbvFlowInvitation.new(nyc_attributes.merge(snap_application_date: "invalid"))
-          expect(invitation).not_to be_valid
-          expect(invitation.errors[:snap_application_date]).to include(
-            I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.snap_application_date.nyc_invalid_date')
-          )
-        end
-
-        it "requires client_id_number" do
-          invitation = CbvFlowInvitation.new(nyc_attributes.merge(client_id_number: nil))
-          expect(invitation).not_to be_valid
-          expect(invitation.errors[:client_id_number]).to include(
-            I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.client_id_number.invalid_format')
-          )
-        end
-      end
-    end
-
-    context "when client_agency_id is 'ma'" do
-      let(:ma_attributes) { valid_attributes.merge(client_agency_id: 'ma') }
-
-      context "user input is invalid" do
-        it "requires agency_id_number" do
-          invitation = CbvFlowInvitation.new(ma_attributes)
-          expect(invitation).not_to be_valid
-          expect(invitation.errors[:agency_id_number]).to include(
-            I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.agency_id_number.invalid_format'),
-          )
-        end
-
-        it "requires beacon_id" do
-          invitation = CbvFlowInvitation.new(ma_attributes)
-          expect(invitation).not_to be_valid
-          expect(invitation.errors[:beacon_id]).to include(
-            I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.beacon_id.invalid_format')
-          )
-        end
-
-        it "requires beacon_id to have 6 alphanumeric characters" do
-          invitation = CbvFlowInvitation.new(ma_attributes.merge(beacon_id: '12345'))
-          expect(invitation).not_to be_valid
-          expect(invitation.errors[:beacon_id]).to include(
-            I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.beacon_id.invalid_format')
-          )
-        end
-
-        it "validates agency_id_number format" do
-          invitation = CbvFlowInvitation.new(ma_attributes.merge(agency_id_number: 'invalid'))
-          expect(invitation).not_to be_valid
-          expect(invitation.errors[:agency_id_number]).to include(
-            I18n.t('activerecord.errors.models.cbv_flow_invitation.attributes.agency_id_number.invalid_format')
-          )
-        end
-
-        it "does not require client_id_number" do
-          invitation = CbvFlowInvitation.new(valid_attributes.merge(client_id_number: nil, client_agency_id: "ma"))
-          expect(invitation).not_to be_valid
-          expect(invitation.errors[:client_id_number]).to be_empty
-        end
-      end
     end
   end
 
diff --git a/app/spec/models/cbv_flow_spec.rb b/app/spec/models/cbv_flow_spec.rb
index fb5b59cf6..d10471385 100644
--- a/app/spec/models/cbv_flow_spec.rb
+++ b/app/spec/models/cbv_flow_spec.rb
@@ -2,11 +2,14 @@
 
 RSpec.describe CbvFlow, type: :model do
   describe ".create_from_invitation" do
-    let(:cbv_flow_invitation) { create(:cbv_flow_invitation, case_number: "ABC1234") }
+    let(:cbv_flow_invitation) { create(:cbv_flow_invitation, cbv_applicant_attributes: { case_number: "ABC1234" }) }
 
     it "copies over relevant fields" do
       cbv_flow = CbvFlow.create_from_invitation(cbv_flow_invitation)
-      expect(cbv_flow).to have_attributes(case_number: "ABC1234", client_agency_id: "sandbox")
+      expect(cbv_flow).to have_attributes(
+        cbv_applicant: cbv_flow_invitation.cbv_applicant,
+        client_agency_id: "sandbox"
+      )
     end
   end
 end
diff --git a/app/spec/services/cbv_invitation_service_spec.rb b/app/spec/services/cbv_invitation_service_spec.rb
index 12ff0bfab..c15dbdb5b 100644
--- a/app/spec/services/cbv_invitation_service_spec.rb
+++ b/app/spec/services/cbv_invitation_service_spec.rb
@@ -3,7 +3,11 @@
 RSpec.describe CbvInvitationService, type: :service do
   let(:event_logger) { instance_double('GenericEventTracker') }
   let(:service) { described_class.new(event_logger) }
-  let(:cbv_flow_invitation_params) { attributes_for(:cbv_flow_invitation) }
+  let(:cbv_flow_invitation_params) do
+    attributes_for(:cbv_flow_invitation).merge(
+      cbv_applicant_attributes: attributes_for(:cbv_applicant)
+    )
+  end
   let(:current_user) { create(:user) }
 
   before do
diff --git a/app/spec/services/data_retention_service_spec.rb b/app/spec/services/data_retention_service_spec.rb
index e21f40548..9ef43960e 100644
--- a/app/spec/services/data_retention_service_spec.rb
+++ b/app/spec/services/data_retention_service_spec.rb
@@ -3,7 +3,7 @@
 RSpec.describe DataRetentionService do
   describe "#redact_invitations" do
     let!(:cbv_flow_invitation) do
-      create(:cbv_flow_invitation)
+      create(:cbv_flow_invitation, :nyc)
     end
     let(:service) { DataRetentionService.new }
     let(:now) { Time.now }
@@ -35,6 +35,14 @@
           )
         end
 
+        it "redacts the associated CbvApplicant" do
+          service.redact_invitations
+          expect(cbv_flow_invitation.cbv_applicant.reload).to have_attributes(
+            case_number: "REDACTED",
+            redacted_at: within(1.second).of(Time.now)
+          )
+        end
+
         it "skips the invitation if it has already been redacted" do
           cbv_flow_invitation.redact!
 
@@ -71,6 +79,11 @@
         expect { service.redact_incomplete_cbv_flows }
           .not_to change { cbv_flow_invitation.reload.attributes }
       end
+
+      it "does not redact the CbvApplicant" do
+        expect { service.redact_incomplete_cbv_flows }
+          .not_to change { cbv_flow.cbv_applicant.reload.attributes }
+      end
     end
 
     context "after the deletion threshold" do
@@ -99,6 +112,13 @@
         )
       end
 
+      it "redacts the associated CbvApplicant" do
+        service.redact_incomplete_cbv_flows
+        expect(cbv_flow.cbv_applicant.reload).to have_attributes(
+          case_number: "REDACTED"
+        )
+      end
+
       it "skips redacting already-redacted CbvFlows" do
         service.redact_incomplete_cbv_flows
 
@@ -187,6 +207,11 @@
         expect { service.redact_complete_cbv_flows }
           .not_to change { cbv_flow_invitation.reload.attributes }
       end
+
+      it "does not redact the CbvApplicant" do
+        expect { service.redact_complete_cbv_flows }
+          .not_to change { cbv_flow.cbv_applicant.reload.attributes }
+      end
     end
 
     context "after the deletion threshold" do
@@ -208,6 +233,13 @@
         )
       end
 
+      it "redacts the associated applicant" do
+        service.redact_complete_cbv_flows
+        expect(cbv_flow.cbv_applicant.reload).to have_attributes(
+          case_number: "REDACTED"
+        )
+      end
+
       it "skips redacting already-redacted CbvFlows" do
         service.redact_complete_cbv_flows
 
@@ -218,9 +250,9 @@
   end
 
   describe ".manually_redact_by_case_number!" do
-    let(:cbv_flow_invitation) { create(:cbv_flow_invitation, case_number: "DELETEME001") }
-    let!(:cbv_flow) { create(:cbv_flow, cbv_flow_invitation: cbv_flow_invitation) }
-    let!(:second_cbv_flow) { create(:cbv_flow, cbv_flow_invitation: cbv_flow_invitation) }
+    let(:cbv_flow_invitation) { create(:cbv_flow_invitation, cbv_applicant_attributes: { case_number: "DELETEME001" }) }
+    let!(:cbv_flow) { CbvFlow.create_from_invitation(cbv_flow_invitation) }
+    let!(:second_cbv_flow) { CbvFlow.create_from_invitation(cbv_flow_invitation) }
 
     it "redacts the invitation and all flow objects" do
       DataRetentionService.manually_redact_by_case_number!("DELETEME001")
@@ -229,6 +261,10 @@
         case_number: "REDACTED",
         redacted_at: within(1.second).of(Time.now)
       )
+      expect(cbv_flow.cbv_applicant.reload).to have_attributes(
+        case_number: "REDACTED",
+        redacted_at: within(1.second).of(Time.now)
+      )
       expect(second_cbv_flow.reload).to have_attributes(
         case_number: "REDACTED",
         redacted_at: within(1.second).of(Time.now)
diff --git a/docs/app/rendered/database-schema.pdf b/docs/app/rendered/database-schema.pdf
index a1fdfdae0..232112ff2 100644
Binary files a/docs/app/rendered/database-schema.pdf and b/docs/app/rendered/database-schema.pdf differ