diff --git a/app/jobs/ptcpnt_persn_id_depnt_org_fix_job.rb b/app/jobs/ptcpnt_persn_id_depnt_org_fix_job.rb
new file mode 100644
index 00000000000..a444566d455
--- /dev/null
+++ b/app/jobs/ptcpnt_persn_id_depnt_org_fix_job.rb
@@ -0,0 +1,55 @@
+# frozen_string_literal: true
+
+require_relative "../../lib/helpers/ptcpnt_persn_id_depnt_org_fix.rb"
+require_relative "../../lib/helpers/master_scheduler_interface.rb"
+class PtcpntPersnIdDepntOrgFixJob < CaseflowJob
+  include MasterSchedulerInterface
+
+  def initialize
+    @stuck_job_report_service = StuckJobReportService.new
+    super
+  end
+
+  def error_text
+    "participantPersonId does not match a dependent or an organization"
+  end
+
+  def perform
+    start_time
+
+    loop_through_and_call_process_records
+
+    end_time
+    log_processing_time
+  end
+
+  def loop_through_and_call_process_records
+    process_records
+  end
+
+  def process_records
+    fix_instance.start_processing_records
+  end
+
+  def records_with_errors
+    fix_instance.class.error_records
+  end
+
+  def log_processing_time
+    (end_time && start_time) ? end_time - start_time : 0
+  end
+
+  def start_time
+    @start_time ||= Time.zone.now
+  end
+
+  def end_time
+    @end_time ||= Time.zone.now
+  end
+
+  private
+
+  def fix_instance
+    @fix_instance ||= PtcpntPersnIdDepntOrgFix.new(@stuck_job_report_service)
+  end
+end
diff --git a/lib/helpers/ptcpnt_persn_id_depnt_org_fix.rb b/lib/helpers/ptcpnt_persn_id_depnt_org_fix.rb
new file mode 100644
index 00000000000..e690d932f92
--- /dev/null
+++ b/lib/helpers/ptcpnt_persn_id_depnt_org_fix.rb
@@ -0,0 +1,168 @@
+# frozen_string_literal: true
+
+# :reek:InstanceVariableAssumption
+class PtcpntPersnIdDepntOrgFix < CaseflowJob
+  ERROR_TEXT = "participantPersonId does not match a dependent or an organization"
+
+  ASSOCIATIONS = [
+    BgsPowerOfAttorney,
+    BgsAttorney,
+    CavcRemandsAppellantSubstitution,
+    Claimant,
+    DecisionIssue,
+    EndProductEstablishment,
+    Notification,
+    Organization,
+    Person,
+    RequestIssue,
+    VbmsDistribution,
+    Veteran
+  ].freeze
+
+  def initialize(stuck_job_report_service)
+    @stuck_job_report_service = stuck_job_report_service
+  end
+
+  def start_processing_records
+    return if self.class.error_records.blank?
+
+    # count of records with errors before fix
+    @stuck_job_report_service.append_record_count(self.class.error_records.count, ERROR_TEXT)
+
+    self.class.error_records.each do |supp_claim|
+      incorrect_pid = supp_claim.claimant.participant_id
+      # check that claimant type is VeteranClaimant
+      next unless supp_claim.claimant.type == "VeteranClaimant"
+
+      veteran_file_number = supp_claim.veteran.file_number
+      @correct_pid = retrieve_correct_pid(veteran_file_number)
+
+      handle_person_and_claimant_records(supp_claim)
+      retrieve_records_to_fix(incorrect_pid)
+
+      @stuck_job_report_service.append_single_record(supp_claim.class.name, supp_claim.id)
+      # Re-run job after fixing broken records
+      re_run_job(supp_claim)
+    end
+    # record count with errors after fix
+    @stuck_job_report_service.append_record_count(self.class.error_records.count, ERROR_TEXT)
+    @stuck_job_report_service.write_log_report(ERROR_TEXT)
+  end
+
+  class << self
+    def iterate_through_associations_with_bad_pid(incorrect_pid, incorrectly_associated_records)
+      ASSOCIATIONS.each do |ass|
+        if ass.attribute_names.include?("participant_id")
+          records = ass.where(participant_id: incorrect_pid)
+          incorrectly_associated_records.push(*records)
+
+        elsif ass.attribute_names.include?("claimant_participant_id")
+          records = ass.where(claimant_participant_id: incorrect_pid)
+          incorrectly_associated_records.push(*records)
+        elsif ass.attribute_names.include?("veteran_participant_id")
+          records = ass.where(veteran_participant_id: incorrect_pid)
+          incorrectly_associated_records.push(*records)
+        end
+      end
+      # Return the updated array
+      incorrectly_associated_records
+    end
+
+    def error_records
+      SupplementalClaim.where("establishment_error ILIKE ?", "%#{ERROR_TEXT}%").where(establishment_canceled_at: nil)
+    end
+end
+
+  private
+
+  def correct_person
+    Person.find_by(participant_id: @correct_pid)
+  end
+
+  def handle_person_and_claimant_records(supp_claim)
+    incorrect_person_record = supp_claim.claimant.person
+
+    ActiveRecord::Base.transaction do
+      if correct_person.present?
+        move_claimants_to_correct_person(correct_person, incorrect_person_record)
+        destroy_incorrect_person_record(incorrect_person_record)
+      else
+        update_incorrect_person_record_participant_id(incorrect_person_record)
+      end
+
+      update_claimant_payee_code(supp_claim.claimant, "00")
+    rescue StandardError => error
+      handle_error(error, supp_claim)
+    end
+  end
+
+  def retrieve_correct_pid(veteran_file_number)
+    begin
+      hash = BGSService.new.fetch_veteran_info(veteran_file_number)
+      hash[:ptcpnt_id]
+    rescue StandardError => error
+      message = "Error retrieving participant ID for veteran file number #{veteran_file_number}: #{error}"
+      @stuck_job_report_service.logs.push(message)
+      log_error(error)
+    end
+  end
+
+  def retrieve_records_to_fix(incorrect_pid)
+    incorrectly_associated_records = self.class.iterate_through_associations_with_bad_pid(incorrect_pid, [])
+
+    incorrectly_associated_records.each do |record|
+      fix_record(record)
+    end
+  end
+
+  def re_run_job(supp_claim)
+    begin
+      DecisionReviewProcessJob.perform_now(supp_claim)
+    rescue StandardError => error
+      @stuck_job_report_service.append_error(supp_claim.class.name, supp_claim.id, error)
+      log_error(error)
+    end
+  end
+
+  def fix_record(record)
+    attribute_name = determine_attribute_name(record)
+    process_record(record, attribute_name)
+  end
+
+  def move_claimants_to_correct_person(correct_person, incorrect_person)
+    correct_person.claimants << incorrect_person.claimants
+    incorrect_person.claimants.clear
+    incorrect_person.save!
+  end
+
+  def destroy_incorrect_person_record(incorrect_person)
+    incorrect_person.destroy!
+  end
+
+  def update_incorrect_person_record_participant_id(incorrect_person)
+    incorrect_person.update(participant_id: @correct_pid)
+  end
+
+  def update_claimant_payee_code(claimant, new_payee_code)
+    claimant.update(payee_code: new_payee_code) if claimant.payee_code != new_payee_code
+  end
+
+  def handle_error(error, record)
+    log_error(error)
+    @stuck_job_report_service.append_error(record.class.name, record.id, error)
+  end
+
+  def determine_attribute_name(record)
+    record.attribute_names.find do |attribute_name|
+      %w[participant_id claimant_participant_id veteran_participant_id].include?(attribute_name)
+    end
+  end
+
+  def process_record(record, attribute_name)
+    ActiveRecord::Base.transaction do
+      record.update(attribute_name => @correct_pid)
+    rescue StandardError => error
+      handle_error(error, record)
+    end
+  end
+end
diff --git a/spec/jobs/ptcpnt_persn_id_depnt_org_fix_job_spec.rb b/spec/jobs/ptcpnt_persn_id_depnt_org_fix_job_spec.rb
new file mode 100644
index 00000000000..aa2396413c9
--- /dev/null
+++ b/spec/jobs/ptcpnt_persn_id_depnt_org_fix_job_spec.rb
@@ -0,0 +1,5 @@
+# frozen_string_literal: true
+
+describe PtcpntPersnIdDepntOrgFixJob, :postgres do
+  it_behaves_like "a Master Scheduler serializable object", PtcpntPersnIdDepntOrgFixJob
+end
diff --git a/spec/lib/helpers/ptcpnt_persn_id_depnt_org_fix_spec.rb b/spec/lib/helpers/ptcpnt_persn_id_depnt_org_fix_spec.rb
new file mode 100644
index 00000000000..34bf8254b87
--- /dev/null
+++ b/spec/lib/helpers/ptcpnt_persn_id_depnt_org_fix_spec.rb
@@ -0,0 +1,181 @@
+# frozen_string_literal: true
+
+require "helpers/ptcpnt_persn_id_depnt_org_fix"
+
+describe PtcpntPersnIdDepntOrgFix, :postgres do
+  let(:error_text) { "participantPersonId does not match a dependent or an organization" }
+  let(:veteran_file_number) { "123456789" }
+  let!(:veteran) { create(:veteran, file_number: veteran_file_number) }
+  let(:correct_pid) { "654321" }
+  let(:incorrect_pid) { "incorrect_pid" }
+  let(:end_product_establishment) do
+    create(
+      :end_product_establishment,
+      claimant_participant_id: incorrect_pid,
+      veteran_file_number: veteran_file_number
+    )
+  end
+  let(:claimant) do
+    create(
+      :claimant,
+      participant_id: incorrect_pid,
+      type: "VeteranClaimant", payee_code: "00"
+    )
+  end
+  let(:correct_person) { create(:person, participant_id: correct_pid, ssn: veteran_file_number) }
+  let!(:supplemental_claim) do
+    create(
+      :supplemental_claim,
+      veteran_file_number: veteran_file_number,
+      establishment_error: error_text,
+      claimants: [
+        claimant
+      ],
+      end_product_establishments: [
+        end_product_establishment
+      ]
+    )
+  end
+  let!(:stuck_job_report_service) { StuckJobReportService.new }
+
+  subject { described_class.new(stuck_job_report_service) }
+
+  describe "BGS Service Call" do
+    context "BGS Service fails" do
+      before do
+        # Stub BGSService to simulate an error response
+        allow_any_instance_of(BGSService).to receive(:fetch_veteran_info).with(veteran_file_number) do
+          fail StandardError, "Simulated BGS error"
+        end
+      end
+      it "handles errors from BGSService and logs errors appropriately" do
+        expect(subject).to receive(:log_error).with(instance_of(StandardError)).at_least(:once)
+
+        subject.send(:retrieve_correct_pid, veteran_file_number)
+      end
+
+      it "logs to the stuck_job_report_service" do
+        logs = subject.instance_variable_get(:@stuck_job_report_service).logs
+        message_1 = "Error retrieving participant ID for veteran file "
+        message_2 = "number #{veteran_file_number}: Simulated BGS error"
+        error_message = message_1 + message_2
+
+        subject.start_processing_records
+        expect(logs).to include(match(/#{error_message}/))
+      end
+    end
+
+    context "BGS Service succeeds" do
+      before do
+        # Stub BGSService to simulate an error response
+        allow_any_instance_of(BGSService).to receive(:fetch_veteran_info) do
+          { ptcpnt_id: correct_pid }
+        end
+      end
+      it "sends back the correct_pid" do
+        subject.start_processing_records
+      end
+    end
+  end
+
+  describe "Claimant is not VeteranClaimant" do
+    it "does NOT remediate the record" do
+      supplemental_claim.claimant.update(type: "DependentClaimant")
+      subject.start_processing_records
+      expect(supplemental_claim.reload.establishment_error).to eq(error_text)
+    end
+  end
+
+  describe "Association Processing" do
+    before do
+      # Stub BGSService to simulate an error response
+      allow_any_instance_of(BGSService).to receive(:fetch_veteran_info) do
+        { ptcpnt_id: correct_pid }
+      end
+    end
+
+    context "BgsPowerOfAttorney record" do
+      let!(:bgs_power_of_attorney) { create(:bgs_power_of_attorney, claimant_participant_id: "incorrect_pid") }
+
+      it "correctly identifies and processes records with incorrect participant_id for BgsPowerOfAttorney" do
+        subject.start_processing_records
+        expect(bgs_power_of_attorney.reload.claimant_participant_id).to eq(correct_pid)
+      end
+    end
+
+    context "EndProductEstablishment record" do
+      let!(:epe) do
+        create(
+          :end_product_establishment,
+          claimant_participant_id: "incorrect_pid",
+          source_id: supplemental_claim.id,
+          source_type: "SupplementalClaim"
+        )
+      end
+
+      it "correctly identifies and processes records with incorrect participant_id for EndProductEstablishment" do
+        subject.start_processing_records
+        expect(epe.reload.claimant_participant_id).to eq(correct_pid)
+      end
+    end
+
+    context "Organization record" do
+      let!(:organization) { create(:organization, participant_id: incorrect_pid) }
+
+      it "correctly identifies and processes records with incorrect participant_id for Organization" do
+        subject.start_processing_records
+        expect(organization.reload.participant_id).to eq(correct_pid)
+      end
+    end
+
+    context "RequestIssue record" do
+      let!(:request_issue) { create(:request_issue, veteran_participant_id: incorrect_pid) }
+
+      it "correctly identifies and processes records with incorrect participant_id for RequestIssue" do
+        subject.start_processing_records
+        expect(request_issue.reload.veteran_participant_id).to eq(correct_pid)
+      end
+    end
+
+    describe "#handle_person_and_claimant_records" do
+      it "handles person records" do
+        correct_person
+        expect do
+          subject.start_processing_records.to
+        end
+        change { Person.count }.by(-1)
+      end
+
+      it "updates supplemental_claim to the correct claimant" do
+        correct_person
+        subject.start_processing_records
+        expect(supplemental_claim.claimant.participant_id).to eq(correct_pid)
+      end
+
+      it "updates supplemental_claim to the correct person" do
+        correct_person
+        subject.start_processing_records
+        expect(supplemental_claim.claimant.person.participant_id).to eq(correct_pid)
+      end
+
+      context "No Person found with correct PID" do
+        it "handles person and claimant records when correct person not found" do
+          expect do
+            subject.start_processing_records
+          end.not_to(change { Person.count })
+        end
+
+        it "updates incorrect person when correct person not found" do
+          subject.start_processing_records
+          expect(supplemental_claim.claimant.person.reload.participant_id).to eq(correct_pid)
+        end
+
+        it "updates payee_code if not 00" do
+          supplemental_claim.claimant.payee_code = nil
+          subject.start_processing_records
+          expect(supplemental_claim.claimant.payee_code).to eq("00")
+        end
+      end
+    end
+  end
+end