AAP-89607: enable controller adoption after restart via receptor work unit reconnection

Add ability for a restarted controller to reconnect to in-progress work units
on the execution environment and replay their events, eliminating the need
to re-run jobs after controller restart in hybrid/mesh deployments.

Key changes:

1. Extract _process_phase() so it can be called standalone without re-transmitting
   artifacts. Unified per-job loops consolidate startup/heartbeat processing
   with explicit adoption/error branches for each job.

2. Add reattach_to_work_unit() to reconnect and stream events from a completed
   receptor work unit. Uses hybrid memory-safe dedup (safe_threshold + collision_zone)
   to avoid re-sending events already in the database to the callback receiver.

3. Unify job finalization: extract _finalize_job_run() shared between normal
   execution and adoption paths. Extract _configure_runner_callback() as the
   canonical callback factory used by both paths.

4. Move _receptor_release_work() after terminal status is committed to DB,
   eliminating the race where a kill between release and commit left jobs
   stuck in 'running' without adoptable work units.

5. Add adopt_job_async @task to run adoption in background, enabling real-time
   event streaming from the EE during controller restart recovery.

Cluster-verified across 9 scenarios with all event stream data checked.
Unit and functional tests cover all adoption paths and edge cases.

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

awx/main/dispatch/reaper.py

@@ -8,24 +8,6 @@ from awx.main.models import Instance, UnifiedJob, WorkflowJob
 logger = logging.getLogger('awx.main.dispatch')
 
 
-def startup_reaping():
-    """
-    If this particular instance is starting, then we know that any running jobs are invalid
-    so we will reap those jobs as a special action here
-    """
-    jobs = UnifiedJob.objects.filter(status='running', controller_node=Instance.objects.my_hostname())
-    job_ids = []
-    for j in jobs:
-        job_ids.append(j.id)
-        reap_job(
-            j,
-            'failed',
-            job_explanation='Task was marked as running at system start up. The system must have not shut down properly, so it has been marked as failed.',
-        )
-    if job_ids:
-        logger.error(f'Unified jobs {job_ids} were reaped on dispatch startup')
-
-
 def reap_job(j, status, job_explanation=None):
     j.refresh_from_db(fields=['status', 'job_explanation'])
     status_before = j.status
@@ -48,8 +30,11 @@ def reap_job(j, status, job_explanation=None):
 
 
 def reap(instance=None, status='failed', job_explanation=None, excluded_uuids=None, ref_time=None):
-    """
-    Reap all jobs in running for this instance.
+    """Reap all running jobs for this instance.
+
+    Per-job adopt-or-reap decisions (based on work_unit_id) belong in the caller's loop,
+    not here as a queryset filter. See _process_startup_jobs and _process_running_jobs
+    in awx.main.tasks.system for the unified loops that replaced undispatched_only filtering.
     """
     if instance is None:
         hostname = Instance.objects.my_hostname()

awx/main/dispatch/worker/callback.py

@@ -195,7 +195,11 @@ class CallbackBrokerWorker:
 
     def flush(self, force=False):
         now = tz_now()
-        if force or (time.time() - self.last_flush) > settings.JOB_EVENT_BUFFER_SECONDS or any([len(events) >= 1000 for events in self.buff.values()]):
+        if (
+            force
+            or (time.time() - self.last_flush) > settings.JOB_EVENT_BUFFER_SECONDS
+            or any(len(events) >= settings.JOB_EVENT_CALLBACK_BUFFER_SIZE for events in self.buff.values())
+        ):
             metrics_bulk_events_saved = 0
             metrics_singular_events_saved = 0
             metrics_events_batch_save_errors = 0

awx/main/tasks/callback.py

@@ -89,6 +89,8 @@ class RunnerCallback:
         self.wrapup_event_dispatched = False
         self.artifacts_processed = False
         self.extra_update_fields = {}
+        self.dedup_threshold = None  # skip events with counter <= this (safe contiguous range already in DB)
+        self.persisted_counters = None  # skip events in this set (collision zone: counters above threshold already in DB)
 
     def update_model(self, pk, _attempt=0, **updates):
         return update_model(self.model, pk, _attempt=0, _max_attempts=self.update_attempts, **updates)
@@ -121,6 +123,57 @@ class RunnerCallback:
             self.delay_update(result_traceback=ANSIBLE_RUNNER_NEEDS_UPDATE_MESSAGE)
         return self.extra_update_fields
 
+    def configure_for_job(self, instance, safe_env=None, dedup_threshold=None, persisted_counters=None):
+        """Initialize callback fields for event processing.
+
+        This is the single source of truth for RunnerCallback initialization, shared
+        between the normal job path (BaseTask.run) and adoption (reattach_to_work_unit).
+        Having both paths call this method means any missing field is caught by the
+        normal test suite rather than only by adoption-specific tests.
+
+        Args:
+            instance: the UnifiedJob model instance being run
+            safe_env: masked environment dict for log output. None → {} (adoption path,
+                where credentials are not available at reconnect time)
+            dedup_threshold: highest counter where all lower counters are in DB (contiguous
+                prefix). event_handler skips counter <= threshold with an O(1) check. None
+                disables dedup (normal jobs, zero overhead).
+            persisted_counters: small set of counters above dedup_threshold that are already
+                in DB (the collision zone from parallel callback worker races). Bounded by
+                worker concurrency, not job size. None when dedup is disabled.
+        """
+        self.instance = instance
+        self.job_created = str(instance.created)  # stamped on every event (callback.py:160)
+        self.safe_env = safe_env if safe_env is not None else {}
+        if getattr(instance, 'spawned_by_workflow', False):
+            try:
+                self.parent_workflow_job_id = instance.get_workflow_job().id
+            except Exception:
+                pass
+        self.dedup_threshold = dedup_threshold
+        self.persisted_counters = persisted_counters
+        # Populate host_map from inventory hostvars using the same logic as normal job path.
+        # This ensures adoption respects inventory types (smart, constructed, normal),
+        # enabled state, and job slicing — the same way write_inventory_file() does.
+        if hasattr(instance, 'inventory') and instance.inventory_id:
+            try:
+                script_params = {"hostvars": True, "towervars": True}
+                if hasattr(instance, 'job_slice_number'):
+                    script_params['slice_number'] = instance.job_slice_number
+                    script_params['slice_count'] = instance.job_slice_count
+                script_data = instance.inventory.get_script_data(**script_params)
+                for hostname, hv in script_data.get('_meta', {}).get('hostvars', {}).items():
+                    self.host_map[hostname] = hv.get('remote_tower_id', '')
+            except Exception:
+                pass  # host_map stays {}; host_id won't be set on replayed events
+
+    @classmethod
+    def create_for_job(cls, instance, safe_env=None, dedup_threshold=None, persisted_counters=None):
+        """Factory: create and configure a RunnerCallback for a job."""
+        callback = cls(model=type(instance))
+        callback.configure_for_job(instance, safe_env, dedup_threshold, persisted_counters)
+        return callback
+
     def event_handler(self, event_data):
         #
         # ⚠️  D-D-D-DANGER ZONE ⚠️
@@ -144,6 +197,13 @@ class RunnerCallback:
         # logger
         if event_data.get('event') == 'keepalive':
             return
+        if self.dedup_threshold is not None:
+            counter = event_data.get('counter')
+            if counter is not None:
+                if counter <= self.dedup_threshold:
+                    return  # contiguous safe range — all in DB, O(1) check
+                if self.persisted_counters and counter in self.persisted_counters:
+                    return  # collision zone — in DB but above threshold, O(small-set) check
 
         if event_data.get(self.event_data_key, None):
             if self.event_data_key != 'job_id':

awx/main/tasks/host_metrics.py

@@ -120,7 +120,6 @@ class HostMetricSummaryMonthlyTask:
         self.records_to_update = []
 
     def execute(self):
-
         with advisory_lock(
             HostMetricSummaryMonthlyTask.LOCK_KEY, lock_session_timeout_milliseconds=HostMetricSummaryMonthlyTask.LOCK_SESSION_TIMEOUT, wait=False
         ) as acquired:

awx/main/tasks/jobs.py

@@ -46,7 +46,6 @@ from awx.main.constants import (
 )
 from awx.main.models import (
     Instance,
-    Inventory,
     InventorySource,
     UnifiedJob,
     Job,
@@ -205,6 +204,42 @@ def dispatch_waiting_jobs(binder):
         UnifiedJob.objects.filter(pk=uj.pk, status='waiting').update(status='running', start_args='')
 
 
+def _finalize_job_run(model, pk, runner_callback, status, extra_fields=None):
+    """Commit terminal status, trigger notifications, and emit websocket status.
+
+    Shared by BaseTask.run() (normal path) and adoption (reattach_to_work_unit path).
+    Uses duck typing on the instance to schedule dependent task/workflow managers.
+    """
+    all_fields = runner_callback.get_delayed_update_fields()
+    if extra_fields:
+        all_fields.update(extra_fields)
+    instance = update_model(model, pk, status=status, select_for_update=True, **all_fields)
+    if not instance:
+        return None
+    if (instance.host_status_counts is not None) or (not runner_callback.wrapup_event_dispatched):
+        events_processed_hook(instance)
+    instance.websocket_emit_status(status)
+
+    # Schedule dependent task/workflow managers via duck typing on the instance.
+    # This works for all job types: Job, ProjectUpdate, InventoryUpdate, etc.
+    instance.log_lifecycle("finalize_run")
+
+    if hasattr(instance, 'unifiedjob_blocked_jobs') and instance.unifiedjob_blocked_jobs.exists():
+        ScheduleTaskManager().schedule()
+
+    if hasattr(instance, 'spawned_by_workflow') and instance.spawned_by_workflow:
+        ScheduleWorkflowManager().schedule()
+
+    # For jobs with inventory, schedule computed fields update (RunJob-specific finalization)
+    if hasattr(instance, 'inventory_id') and instance.inventory_id:
+        try:
+            update_inventory_computed_fields.delay(instance.inventory_id)
+        except Exception:
+            logger.exception(f'{instance.log_format} Error scheduling inventory computed fields update')
+
+    return instance
+
+
 class BaseTask(object):
     model = None
     event_model = None
@@ -486,15 +521,11 @@ class BaseTask(object):
 
     def write_inventory_file(self, inventory, private_data_dir, file_name, script_params):
         script_data = inventory.get_script_data(**script_params)
-        for hostname, hv in script_data.get('_meta', {}).get('hostvars', {}).items():
-            # maintain a list of host_name --> host_id
-            # so we can associate emitted events to Host objects
-            self.runner_callback.host_map[hostname] = hv.get('remote_tower_id', '')
         file_content = '#! /usr/bin/env python3\n# -*- coding: utf-8 -*-\nprint(%r)\n' % json.dumps(script_data)
         return self.write_private_data_file(private_data_dir, file_name, file_content, sub_dir='inventory', file_permissions=0o700)
 
     def build_inventory(self, instance, private_data_dir):
-        script_params = dict(hostvars=True, towervars=True)
+        script_params = {"hostvars": True, "towervars": True}
         if hasattr(instance, 'job_slice_number'):
             script_params['slice_number'] = instance.job_slice_number
             script_params['slice_count'] = instance.job_slice_count
@@ -612,18 +643,6 @@ class BaseTask(object):
         """
         instance.log_lifecycle("post_run")
 
-    def final_run_hook(self, instance, status, private_data_dir):
-        """
-        Hook for any steps to run after job/task is marked as complete.
-        """
-        instance.log_lifecycle("finalize_run")
-
-        # Run task manager appropriately for speculative dependencies
-        if instance.unifiedjob_blocked_jobs.exists():
-            ScheduleTaskManager().schedule()
-        if instance.spawned_by_workflow:
-            ScheduleWorkflowManager().schedule()
-
     def should_use_fact_cache(self):
         return False
 
@@ -663,6 +682,7 @@ class BaseTask(object):
 
         self.safe_cred_env = {}
         private_data_dir = None
+        receptor_job = None
 
         try:
             if self.instance.execution_environment_id is None:
@@ -703,16 +723,9 @@ class BaseTask(object):
             self.build_extra_vars_file(self.instance, private_data_dir)
             args = self.build_args(self.instance, private_data_dir, passwords)
             env = self.build_env(self.instance, private_data_dir, private_data_files=private_data_files)
-            self.runner_callback.safe_env = build_safe_env(env)
-
-            self.runner_callback.instance = self.instance
-
-            # store a reference to the parent workflow job (if any) so we can include
-            # it in event data JSON
-            if self.instance.spawned_by_workflow:
-                self.runner_callback.parent_workflow_job_id = self.instance.get_workflow_job().id
-
-            self.runner_callback.job_created = str(self.instance.created)
+            # Initialize common callback fields so that the normal job path exercises
+            # the same configuration code as adoption.
+            self.runner_callback.configure_for_job(self.instance, safe_env=build_safe_env(env))
 
             credentials = self._credentials
 
@@ -795,6 +808,13 @@ class BaseTask(object):
                 self.unit_id = receptor_job.unit_id
 
                 if not res:
+                    # res is None when quota exceeded or other early-return condition.
+                    # Must release work unit here before returning, or it will leak.
+                    if receptor_job and getattr(receptor_job, 'receptor_ctl', None):
+                        try:
+                            receptor_job._receptor_release_work(receptor_job.receptor_ctl, 'error')
+                        except Exception:
+                            logger.exception(f'Failed to release work unit {getattr(receptor_job, "unit_id", "unknown")} on early return')
                     return
 
             status = res.status
@@ -832,23 +852,21 @@ class BaseTask(object):
         except Exception:
             logger.exception('{} Post run hook errored.'.format(self.instance.log_format))
 
-        self.instance = self.update_model(pk)
-        self.instance = self.update_model(pk, status=status, select_for_update=True, **self.runner_callback.get_delayed_update_fields())
+        self.private_data_dir = private_data_dir
+        try:
+            self.instance = _finalize_job_run(self.model, pk, self.runner_callback, status)
+        finally:
+            # Guarantee work unit release even if finalization throws
+            if receptor_job and getattr(receptor_job, 'receptor_ctl', None):
+                try:
+                    receptor_job._receptor_release_work(receptor_job.receptor_ctl, status)
+                except Exception:
+                    logger.exception(f'Failed to release work unit {getattr(receptor_job, "unit_id", "unknown")}')
 
-        # Field host_status_counts is used as a metric to check if event processing is finished
-        # we send notifications if it is, if not, callback receiver will send them
         if not self.instance:
             logger.error(f'Unified job pk={pk} appears to be deleted while running')
             return
-        if (self.instance.host_status_counts is not None) or (not self.runner_callback.wrapup_event_dispatched):
-            events_processed_hook(self.instance)
 
-        try:
-            self.final_run_hook(self.instance, status, private_data_dir)
-        except Exception:
-            logger.exception('{} Final run hook errored.'.format(self.instance.log_format))
-
-        self.instance.websocket_emit_status(status)
         if status != 'successful':
             if status == 'canceled':
                 raise AwxTaskError.TaskCancel(self.instance, rc)
@@ -1374,16 +1392,6 @@ class RunJob(SourceControlMixin, BaseTask):
                 job_created=job.created,
             )
 
-    def final_run_hook(self, job, status, private_data_dir):
-        super(RunJob, self).final_run_hook(job, status, private_data_dir)
-        try:
-            inventory = job.inventory
-        except Inventory.DoesNotExist:
-            pass
-        else:
-            if inventory is not None:
-                update_inventory_computed_fields.delay(inventory.id)
-
 
 @task(queue=get_task_queuename)
 class RunProjectUpdate(BaseTask):
@@ -1826,7 +1834,7 @@ class RunInventoryUpdate(SourceControlMixin, BaseTask):
             inventory_update.log_lifecycle("start_job_fact_cache")
             for input_inventory in inventory_update.inventory.input_inventories.all():
                 args.append('-i')
-                script_params = dict(hostvars=True, towervars=True)
+                script_params = {"hostvars": True, "towervars": True}
                 source_inv_path = self.write_inventory_file(input_inventory, private_data_dir, f'hosts_{input_inventory.id}', script_params)
                 args.append(get_incontainer_path(source_inv_path, private_data_dir))
                 # Include any facts from input inventories so they can be used in filters

awx/main/tasks/receptor.py

@@ -3,16 +3,20 @@ from base64 import b64encode
 from collections import namedtuple
 import concurrent.futures
 from enum import Enum
+import json
 import logging
 import os
 import shutil
 import socket
+import tempfile
 import time
 import yaml
 
 # Django
 from django.conf import settings
 from django.db import connections
+from django.db.models import Exists, OuterRef
+from django.utils.timezone import now
 
 # Runner
 import ansible_runner
@@ -31,11 +35,13 @@ from awx.main.utils.common import (
     parse_yaml_or_json,
     cleanup_new_process,
 )
-from awx.main.constants import MAX_ISOLATED_PATH_COLON_DELIMITER
+from awx.main.constants import JOB_FOLDER_PREFIX, MAX_ISOLATED_PATH_COLON_DELIMITER
 from awx.main.tasks.signals import signal_state, signal_callback, SignalExit
+from awx.main.tasks.callback import RunnerCallback
 from awx.main.models import Instance, InstanceLink, UnifiedJob, ReceptorAddress
 from awx.main.dispatch import get_task_queuename
 
+
 # Receptorctl
 from receptorctl.socket_interface import ReceptorControl
 
@@ -175,6 +181,37 @@ def get_receptor_ctl(config_data=None):
         return ReceptorControl(receptor_sockfile)
 
 
+def adopt_remote_work(receptor_ctl, node, unit_id, config_data=None):
+    """Adopt a running work unit from a remote node, passing TLS/signwork when configured.
+
+    PR#1564 adds adopt_work() to receptorctl. If not available (older client), falls back
+    to the JSON control-socket protocol directly — compatible with any PR#1564 receptor server.
+
+    TEMPORARY: Once receptor#1564 becomes the minimum required version and requirements.txt
+    is updated, the fallback block (lines ~197–206) can be removed since adopt_work() will
+    always be available. This handles mixed-version deployments during transition.
+    """
+    if config_data is None:
+        config_data = read_receptor_config()
+    tls_client = get_tls_client(config_data, True)
+    sign = work_signing_enabled(config_data)
+
+    if hasattr(receptor_ctl, 'adopt_work'):
+        return receptor_ctl.adopt_work(node, unit_id, tlsclient=tls_client or None, signwork=sign)
+
+    # TODO: Delete this fallback block once receptor#1564 is the minimum required version.
+    # Older receptorctl: send JSON directly so the server's InitFromJSON path handles TLS/signwork.
+    # The text-protocol InitFromString path does NOT support tlsclient/signwork.
+    command = {"command": "work", "subcommand": "adopt", "node": node, "unitid": unit_id}
+    if tls_client:
+        command["tlsclient"] = tls_client
+    if sign:
+        command["signwork"] = True
+    receptor_ctl.connect()
+    receptor_ctl.writestr(json.dumps(command) + "\n")
+    return receptor_ctl.read_and_parse_json()
+
+
 def find_node_in_mesh(node_name, receptor_ctl):
     attempts = 10
     backoff = 1
@@ -406,49 +443,41 @@ class AWXReceptorJob:
     def run(self):
         # We establish a connection to the Receptor socket
         self.config_data = read_receptor_config()
-        receptor_ctl = get_receptor_ctl(self.config_data)
-
-        res = None
-        try:
-            res = self._run_internal(receptor_ctl)
-            return res
-        finally:
-            status = getattr(res, 'status', 'error')
-            self._receptor_release_work(receptor_ctl, status)
+        self.receptor_ctl = get_receptor_ctl(self.config_data)
+        return self._run_internal(self.receptor_ctl)
 
     def _receptor_release_work(self, receptor_ctl: ReceptorControl, status: str) -> None:
-        """
-        Releases the work unit from Receptor if certain conditions are met.
-        This method checks several conditions before attempting to release the work unit:
-        - If `self.unit_id` is `None`, the method returns immediately.
-        - If the `RECEPTOR_RELEASE_WORK` setting is `False`, the method returns immediately.
-        - If the `RECEPTOR_KEEP_WORK_ON_ERROR` setting is `True` and the status is 'error', the method returns immediately.
-        If none of the above conditions are met, the method attempts to release the work unit using the Receptor control command.
-        If an exception occurs during the release process, it logs an error message.
-        Args:
-            receptor_ctl (ReceptorControl): The Receptor control object used to issue commands.
-            status (str): The status of the work unit, which may affect whether it is released.
-        """
-
         if self.unit_id is None:
-            logger.debug("No work unit ID to release.")
             return
 
         if settings.RECEPTOR_RELEASE_WORK is False:
-            logger.debug(f"RECEPTOR_RELEASE_WORK is False, not releasing work unit {self.unit_id}.")
             return
 
         if settings.RECEPTOR_KEEP_WORK_ON_ERROR and status == 'error':
-            logger.debug(f"RECEPTOR_KEEP_WORK_ON_ERROR is True and status is 'error', not releasing work unit {self.unit_id}.")
             return
 
         try:
-            logger.debug(f"Released work unit {self.unit_id}.")
             receptor_ctl.simple_command(f"work release {self.unit_id}")
+            logger.debug(f"Released work unit {self.unit_id}.")
         except Exception:
             logger.exception(f"Error releasing work unit {self.unit_id}.")
 
     def _run_internal(self, receptor_ctl):
+        self._transmit_phase(receptor_ctl)
+        return self._process_phase(receptor_ctl)
+
+    def _transmit_phase(self, receptor_ctl):
+        """Submit work to receptor and wait for the transmit thread to finish.
+
+        Creates the receptor work unit, streams the job payload (private_data_dir + kwargs)
+        to the EE via a socketpair, and saves the resulting unit_id to the DB.
+
+        After this returns, artifacts/ is deleted from private_data_dir. Fact-cache and
+        other input artifacts are transmitted to the EE during this phase; by the time
+        _process_phase() reads artifacts output, the dir is empty so there is no stale
+        input state. This means _process_phase() always receives a clean private_data_dir
+        whether called from _run_internal() or from reattach_to_work_unit().
+        """
         # Create a socketpair. Where the left side will be used for writing our payload
         # (private data dir, kwargs). The right side will be passed to Receptor for
         # reading.
@@ -497,7 +526,20 @@ class AWXReceptorJob:
         if self.work_type != 'local' and os.path.exists(artifact_dir):
             shutil.rmtree(artifact_dir)
 
+    def _process_phase(self, receptor_ctl):
+        """Stream events from the receptor work unit via the ansible-runner process streamer.
+
+        Extracted from _run_internal() so it can be called standalone for job adoption
+        after a same-controller restart (reattach_to_work_unit). The transmit phase
+        (submit_work + transmit) is not repeated — the EE is already running.
+        """
+        resultsock = None
+        resultfile = None
+        try:
             resultsock, resultfile = receptor_ctl.get_work_results(self.unit_id, return_socket=True, return_sockfile=True)
+        except Exception:
+            logger.exception(f'Failed to get work results for unit {self.unit_id}')
+            raise
 
         connections.close_all()
 
@@ -516,7 +558,12 @@ class AWXReceptorJob:
                 res = processor_future.result()
             except SignalExit:
                 receptor_ctl.simple_command(f"work cancel {self.unit_id}")
+                if resultsock:
+                    try:
                         resultsock.shutdown(socket.SHUT_RDWR)
+                    except Exception:
+                        pass
+                if resultfile:
                     resultfile.close()
                 result = namedtuple('result', ['status', 'rc'])
                 res = result('canceled', 1)
@@ -524,6 +571,12 @@ class AWXReceptorJob:
                 signal_state.raise_exception = False
 
             if res.status == 'error':
+                return self._handle_work_error(receptor_ctl, res)
+
+        return res
+
+    def _handle_work_error(self, receptor_ctl, res):
+        """Handle the error path from _process_phase when the work unit status is 'error'."""
         # If ansible-runner ran, but an error occured at runtime, the traceback information
         # is saved via the status_handler passed in to the processor.
         if 'result_traceback' in self.task.runner_callback.extra_update_fields:
@@ -531,7 +584,7 @@ class AWXReceptorJob:
 
         try:
             unit_status = receptor_ctl.simple_command(f'work status {self.unit_id}')
-                    detail = unit_status.get('Detail', None)
+            detail = unit_status.get('Detail') or ''
             state_name = unit_status.get('StateName', None)
             stdout_size = unit_status.get('StdoutSize', 0)
         except Exception:
@@ -545,7 +598,7 @@ class AWXReceptorJob:
             log_name = self.task.instance.log_format
             logger.warning(f"Could not launch pod for {log_name}. Exceeded quota.")
             self.task.update_model(self.task.instance.pk, status='pending')
-                    return
+            return None
 
         try:
             receptor_output = ''
@@ -554,9 +607,10 @@ class AWXReceptorJob:
                 # contain useful information about why the job failed. In case stdout is
                 # massive, only ask for last 1000 bytes
                 startpos = max(stdout_size - 1000, 0)
-                        resultsock, resultfile = receptor_ctl.get_work_results(self.unit_id, startpos=startpos, return_socket=True, return_sockfile=True)
+                _resultsock, resultfile = receptor_ctl.get_work_results(self.unit_id, startpos=startpos, return_socket=True, return_sockfile=True)
                 lines = resultfile.readlines()
                 receptor_output = b"".join(lines).decode()
+                _resultsock.shutdown(socket.SHUT_RDWR)
             if receptor_output:
                 self.task.runner_callback.delay_update(result_traceback=f'Worker output:\n{receptor_output}')
             elif detail:
@@ -799,6 +853,199 @@ RECEPTOR_CONFIG_STARTER = (
 )
 
 
+class _AdoptionTask:
+    """Minimal task stub passed to AWXReceptorJob when called from reattach_to_work_unit.
+
+    Stands in for BaseTask without re-executing task-manager logic. update_model is a
+    no-op since adoption doesn't touch scheduling fields (those are handled by the
+    reaper/heartbeat path that triggered the adoption).
+    """
+
+    def __init__(self, instance, runner_callback):
+        self.instance = instance
+        self.runner_callback = runner_callback
+        self.private_data_dir = None  # Set by reattach_to_work_unit before finalization
+
+    def build_execution_environment_params(self, _instance, _private_data_dir):
+        return {}
+
+    def update_model(self, pk, **kwargs):
+        # Intentionally a no-op: adoption reuses _process_phase without task-manager field updates.
+        # Heartbeat/reaper handles scheduling state; no fields need updating here.
+        pass
+
+
+def _get_adoption_exit_code(unit_status, state_name):
+    """Extract exit code from a finished receptor work unit status dict."""
+    if 'ExitCode' in unit_status:
+        return unit_status['ExitCode']
+    detail = unit_status.get('Detail', '')
+    try:
+        return int(detail.split()[-1])
+    except (ValueError, IndexError):
+        return 0 if state_name == 'Succeeded' else 1
+
+
+def _build_adoption_callback(job, dedup_threshold, collision_zone):
+    """Construct a RunnerCallback for event replay during adoption."""
+    return RunnerCallback.create_for_job(job, dedup_threshold=dedup_threshold, persisted_counters=collision_zone)
+
+
+def _get_or_create_private_data_dir(job):
+    """Return a writable private_data_dir for the adoption process streamer.
+
+    On same-controller restart the original tempdir may be gone (e.g. on OCP where /tmp
+    is wiped on pod restart). ansible-runner's process streamer only needs a writable root
+    directory — it creates artifacts/job_events/ itself.
+
+    Uses AWX_ISOLATION_BASE_PATH (the same root as all other job working dirs) so that
+    sysadmin-configured isolation paths (e.g. persistent volumes on OCP) are respected.
+    """
+    return tempfile.mkdtemp(
+        prefix=JOB_FOLDER_PREFIX % job.pk + 'adoption_',
+        dir=settings.AWX_ISOLATION_BASE_PATH,
+    )
+
+
+def _finalize_adopted_job(job, callback, exit_code, process_phase_failed):
+    """Commit terminal status for an adopted job via the shared _finalize_job_run path.
+
+    The shared finalization function uses duck typing to schedule task/workflow managers
+    for speculative dependencies and workflow jobs.
+
+    Guards before calling: if awx_receptor_workunit_reaper already committed the final
+    status, there is nothing left to do.
+    """
+    from awx.main.tasks.jobs import _finalize_job_run
+
+    job.refresh_from_db(fields=['status'])
+    if job.status != 'running':
+        return
+
+    final_status = 'successful' if exit_code == 0 else 'failed'
+    finished_at = now()
+    extra = {'finished': finished_at}
+    if job.started:
+        extra['elapsed'] = (finished_at - job.started).total_seconds()
+
+    _finalize_job_run(type(job), job.pk, callback, final_status, extra_fields=extra)
+
+    label = 'exit_code (process phase raised)' if process_phase_failed else 'adoption'
+    logger.info(f'Job {job.id} finalized via {label}: {final_status}')
+
+
+def _compute_adoption_dedup(job):
+    """Return (safe_threshold, collision_zone) for counter-skip dedup during adoption.
+
+    Hybrid approach — memory is O(1) + O(worker_count), never O(total events):
+
+    safe_threshold: highest counter where all lower counters are also in DB (contiguous
+        prefix). Events <= this are skipped with a single integer comparison.
+
+    collision_zone: small set of counters above safe_threshold that ARE in DB. These
+        exist because parallel callback workers can commit a higher-counter event before
+        a lower-counter one. Bounded by JOB_EVENT_WORKERS × batch size, typically < 20
+        regardless of total job event count.
+    """
+    next_ctr = job.get_event_queryset().filter(counter=OuterRef('counter') + 1)
+    gap_event = job.get_event_queryset().annotate(has_next=Exists(next_ctr)).filter(has_next=False).order_by('counter').first()
+    safe_threshold = gap_event.counter if gap_event else 0
+
+    cap = settings.JOB_EVENT_WORKERS * settings.JOB_EVENT_CALLBACK_BUFFER_SIZE
+    collision_zone_list = list(job.get_event_queryset().filter(counter__gt=safe_threshold).values_list('counter', flat=True))
+
+    if len(collision_zone_list) > cap:
+        logger.warning(
+            f'Job {job.id}: collision_zone has {len(collision_zone_list)} events above safe_threshold, '
+            f'exceeds dedup cap of {cap}. Events beyond cap may be re-processed if replayed. '
+            f'Consider increasing JOB_EVENT_CALLBACK_BUFFER_SIZE or reducing parallel callback workers.'
+        )
+
+    collision_zone = set(collision_zone_list[:cap])
+    return safe_threshold, collision_zone
+
+
+def reattach_to_work_unit(job, receptor_ctl):
+    """Reconnect to a receptor work unit and stream events in real-time until it completes.
+
+    Reconstructs the minimal process-phase context from the DB job record, then calls
+    _process_phase which blocks until the work unit finishes — streaming events live as
+    the EE generates them. Dedup (safe_threshold + collision_zone) skips events already
+    in DB so replay from startpos=0 is safe.
+
+    Intended to be called from adopt_job_async (a background task) so the caller is not
+    blocked. Cross-controller path (node=execution_node) is deferred to AAP-89602.
+    """
+    unit_id = job.work_unit_id
+
+    # Check state for logging — no longer a gate. We stream regardless.
+    try:
+        unit_status = receptor_ctl.simple_command(f'work status {unit_id}')
+    except Exception:
+        logger.warning(f'Cannot get receptor status for work unit {unit_id} (job {job.id}), deferring adoption')
+        return False
+
+    state_name = unit_status.get('StateName', '')
+    logger.info(f'Adopting job {job.id}: unit {unit_id} in state {state_name!r}, starting real-time streaming')
+
+    # Pending/Running — EE not yet finished; defer to next heartbeat.
+    if state_name in ('Pending', 'Running'):
+        logger.info(f'Job {job.id}: unit {unit_id} in state {state_name!r}, deferring to next heartbeat')
+        return False
+
+    safe_threshold, collision_zone = _compute_adoption_dedup(job)
+    max_counter = max(collision_zone) if collision_zone else safe_threshold
+    logger.info(
+        f'Job {job.id}: safe_threshold={safe_threshold} collision_zone_size={len(collision_zone)} '
+        f'(max counter={max_counter}), replaying from startpos=0 with counter-skip'
+    )
+
+    callback = _build_adoption_callback(job, safe_threshold, collision_zone)
+
+    private_data_dir = _get_or_create_private_data_dir(job)
+    adoption_task = _AdoptionTask(job, callback)
+    adoption_task.private_data_dir = private_data_dir  # Available to final_run_hook
+    receptor_job = AWXReceptorJob(adoption_task, {'private_data_dir': private_data_dir})
+    receptor_job.unit_id = unit_id
+
+    process_phase_failed = False
+    res = None
+    try:
+        res = receptor_job._process_phase(receptor_ctl)  # blocks until unit completes
+    except Exception:
+        logger.exception(f'Adoption process phase failed for job {job.id} (unit={unit_id})')
+        process_phase_failed = True
+
+    # Finalize status to DB first, then release the work unit — matching the ordering
+    # in BaseTask.run() where release follows _finalize_job_run.
+    # Keep private_data_dir available through finalization for final_run_hook.
+    #
+    # _process_phase -> _handle_work_error may return None for 'exceeded quota' where
+    # the job is already set to 'pending' — don't finalize, let the next dispatch handle it.
+    try:
+        if res is not None:
+            exit_code = 0 if getattr(res, 'status', '') == 'successful' else 1
+            _finalize_adopted_job(job, callback, exit_code, process_phase_failed)
+        elif not process_phase_failed:
+            logger.info(f'Job {job.id}: adoption deferred (handled in _handle_work_error)')
+        else:
+            # _process_phase raised — fall back to work status for exit code
+            try:
+                unit_status = receptor_ctl.simple_command(f'work status {unit_id}')
+                state_name = unit_status.get('StateName', '')
+            except Exception:
+                logger.debug(f'Could not get final status for {unit_id} after process phase failure')
+                unit_status = {}
+                state_name = ''
+            exit_code = _get_adoption_exit_code(unit_status, state_name)
+            _finalize_adopted_job(job, callback, exit_code, process_phase_failed)
+    finally:
+        receptor_job._receptor_release_work(receptor_ctl, getattr(res, 'status', 'error'))
+        shutil.rmtree(private_data_dir, ignore_errors=True)
+
+    return True
+
+
 def should_update_config(new_config):
     '''
     checks that the list of instances matches the list of

awx/main/tasks/system.py

@@ -36,7 +36,9 @@ from django.conf import settings
 from django.contrib.auth.models import User
 from django.core.cache import cache
 from django.core.exceptions import ObjectDoesNotExist
+from django.contrib.contenttypes.models import ContentType
 from django.db import DatabaseError, IntegrityError, connection, transaction
+from django.db.models import Max, Q
 from django.db.models.fields.related import ForeignKey
 from django.db.models.query import QuerySet
 from django.utils.encoding import smart_str
@@ -65,6 +67,7 @@ from awx.main.models import (
     SmartInventoryMembership,
     TowerScheduleState,
     UnifiedJob,
+    WorkflowJob,
     convert_jsonfields,
 )
 from awx.main.models.credential import CredentialType
@@ -73,6 +76,7 @@ from awx.main.tasks.host_indirect import save_indirect_host_entries
 from awx.main.tasks.receptor import (
     administrative_workunit_reaper,
     get_receptor_ctl,
+    reattach_to_work_unit,
     worker_cleanup,
     worker_info,
     write_receptor_config,
@@ -149,7 +153,10 @@ def _run_dispatch_startup_common():
     #
     apply_cluster_membership_policies()
     cluster_node_heartbeat(None)
-    reaper.startup_reaping()
+    # Safety net: reap undispatched startup jobs even if cluster_node_heartbeat returned early
+    # (receptor unavailable, rejoining cluster). _process_startup_jobs handles adoption; this
+    # handles the reap-only case. reap_job() is idempotent — already-reaped jobs are skipped.
+    _startup_reap_undispatched(settings.CLUSTER_HOST_ID)
     m = DispatcherMetrics()
     m.reset_values()
 
@@ -661,8 +668,9 @@ def cluster_node_heartbeat(binder):
     Uses Control API to get running tasks.
     """
 
-    # Run common instance management logic
-    this_inst, instance_list, lost_instances = _heartbeat_instance_management()
+    # Run common instance management logic — ctl is the same receptor connection used for
+    # mesh status; we reuse it for the job processing loop to avoid a second socket open.
+    this_inst, instance_list, lost_instances, _ctl = _heartbeat_instance_management()
     if this_inst is None:
         return  # Early return case from instance management
 
@@ -672,19 +680,24 @@ def cluster_node_heartbeat(binder):
     # Handle lost instances
     _heartbeat_handle_lost_instances(lost_instances, this_inst)
 
-    # Get running tasks using dispatcherd API
     if binder is None:
+        # Startup: one loop over all running jobs on this instance.
+        # Dispatched jobs are handed to adopt_job_async; undispatched jobs are reaped.
+        _process_startup_jobs(this_inst)
         logger.debug("Heartbeat finished in startup.")
         return
+
+    # Periodic heartbeat: get running tasks from dispatcherd, then process orphaned jobs.
     active_task_ids = _get_active_task_ids_from_dispatcherd(binder)
     if active_task_ids is None:
         logger.warning("No active task IDs retrieved from dispatcherd, skipping reaper")
         return  # Failed to get task IDs, don't attempt reaping
 
-    # Run local reaper using tasks from dispatcherd
-    ref_time = now()  # No dispatch_time in dispatcherd version
-    logger.debug(f"Running reaper with {len(active_task_ids)} excluded UUIDs")
-    reaper.reap(instance=this_inst, excluded_uuids=active_task_ids, ref_time=ref_time)
+    # One loop over all orphaned running jobs — adopt dispatched, reap undispatched.
+    ref_time = now()
+    logger.debug(f"Running job processing loop with {len(active_task_ids)} excluded UUIDs")
+    _process_running_jobs(this_inst, active_task_ids, ref_time)
+
     # If waiting jobs are hanging out, resubmit them
     if UnifiedJob.objects.filter(controller_node=settings.CLUSTER_HOST_ID, status='waiting').exists():
         from awx.main.tasks.jobs import dispatch_waiting_jobs
@@ -772,7 +785,7 @@ def _heartbeat_instance_management():
         if this_inst:
             this_inst.local_health_check()
             this_inst.mark_offline(errors='Receptor not available')
-        return None, None, None
+        return None, None, None, None
 
     try:
         mesh_status = ctl.simple_command('status')
@@ -795,7 +808,7 @@ def _heartbeat_instance_management():
         this_inst.local_health_check()
         if startup_event and this_inst.capacity != 0:
             logger.warning(f'Rejoining the cluster as instance {this_inst.hostname}. Prior last_seen {last_last_seen}')
-            return None, None, None  # Early return case
+            return None, None, None, None  # Early return case
         elif not last_last_seen:
             logger.warning(f'Instance does not have recorded last_seen, updating to {nowtime}')
         elif (nowtime - last_last_seen) > timedelta(seconds=settings.CLUSTER_NODE_HEARTBEAT_PERIOD + 2):
@@ -808,15 +821,15 @@ def _heartbeat_instance_management():
             this_inst.local_health_check()
         else:
             logger.error("Cluster Host Not Found: {}".format(settings.CLUSTER_HOST_ID))
-            return None, None, None
+            return None, None, None, None
 
     if lost_instances and not _mesh_all_ready_nodes_visible(mesh_status):
         # Mesh gate blocks cleanup, but execution and hop nodes can still be reaped
         # (they don't depend on mesh consensus). Defer only control nodes.
         execution_hop_lost = [inst for inst in lost_instances if inst.node_type in ('execution', 'hop')]
-        return this_inst, instance_list, execution_hop_lost
+        return this_inst, instance_list, execution_hop_lost, ctl
 
-    return this_inst, instance_list, lost_instances
+    return this_inst, instance_list, lost_instances, ctl
 
 
 def _heartbeat_check_versions(this_inst, instance_list):
@@ -841,9 +854,17 @@ def _heartbeat_check_versions(this_inst, instance_list):
 def _reap_and_mark_lost_instance(other_inst):
     """Reap a lost instance's running jobs and mark it offline (or deprovision it)."""
     try:
-        # Only reap jobs never dispatched to receptor; dispatched jobs are left for adoption (AAP-89602)
-        explanation = "Job reaped due to instance shutdown"
-        reaper.reap(other_inst, job_explanation=explanation)
+        workflow_ctype_id = ContentType.objects.get_for_model(WorkflowJob).id
+        running_jobs = list(
+            UnifiedJob.objects.filter(
+                Q(execution_node=other_inst.hostname) | Q(controller_node=other_inst.hostname),
+                status='running',
+            ).exclude(polymorphic_ctype_id=workflow_ctype_id)
+        )
+        for j in running_jobs:
+            # AAP-89602: cross-controller adoption for dispatched jobs will be added here
+            # once ansible/receptor#1564 merges. Both paths reap identically for now.
+            reaper.reap_job(j, 'failed', job_explanation='Job reaped due to instance shutdown')
         # Any jobs that were waiting to be processed by this node will be handed back to task manager
         UnifiedJob.objects.filter(status='waiting', controller_node=other_inst.hostname).update(status='pending', controller_node='', execution_node='')
     except Exception:
@@ -884,6 +905,135 @@ def _heartbeat_handle_lost_instances(lost_instances, this_inst):
             _reap_and_mark_lost_instance(other_inst)
 
 
+def _startup_reap_undispatched(hostname):
+    """Reap undispatched running jobs at startup using only the hostname — no receptor needed.
+
+    Called unconditionally from _run_dispatch_startup_common so undispatched jobs are always
+    reaped, even when cluster_node_heartbeat() returns early (receptor unavailable, rejoining
+    cluster). reap_job() is idempotent: jobs already handled by _process_startup_jobs are
+    skipped via the running-status check.
+    """
+    workflow_ctype_id = ContentType.objects.get_for_model(WorkflowJob).id
+    jobs = list(
+        UnifiedJob.objects.filter(
+            status='running',
+            controller_node=hostname,
+        )
+        .filter(Q(work_unit_id='') | Q(work_unit_id=None))
+        .exclude(polymorphic_ctype_id=workflow_ctype_id)
+    )
+    job_ids = [j.id for j in jobs]
+    for j in jobs:
+        reaper.reap_job(
+            j,
+            'failed',
+            job_explanation='Task was marked as running at system start up. The system must have not shut down properly, so it has been marked as failed.',
+        )
+    if job_ids:
+        logger.error(f'Unified jobs {job_ids} were reaped on dispatch startup')
+
+
+def _process_startup_jobs(this_inst):
+    """On controller restart, process all running jobs owned by this instance in one loop.
+
+    Dispatched jobs (work_unit_id set) are handed off to adopt_job_async, which streams
+    events in real-time in a background task. Undispatched jobs are reaped immediately.
+
+    Same-controller only. Cross-controller adoption (jobs from a dead peer controller) is
+    deferred to AAP-89602 and requires ansible/receptor#1564.
+    """
+    workflow_ctype_id = ContentType.objects.get_for_model(WorkflowJob).id
+    jobs = list(UnifiedJob.objects.filter(status='running', controller_node=this_inst.hostname).exclude(polymorphic_ctype_id=workflow_ctype_id))
+    if not jobs:
+        return
+
+    reaped_ids = []
+    for j in jobs:
+        try:
+            if j.work_unit_id:
+                obj, _ = adopt_job_async.apply_async(args=[j.id], queue=get_task_queuename())
+                UnifiedJob.objects.filter(pk=j.id).update(celery_task_id=obj['uuid'])
+            else:
+                reaped_ids.append(j.id)
+                reaper.reap_job(
+                    j,
+                    'failed',
+                    job_explanation='Task was marked as running at system start up. The system must have not shut down properly, so it has been marked as failed.',
+                )
+        except Exception:
+            logger.exception(f'Failed processing job {j.id} in startup job loop')
+    if reaped_ids:
+        logger.error(f'Unified jobs {reaped_ids} were reaped on dispatch startup')
+
+
+def _process_running_jobs(this_inst, active_task_ids, ref_time):
+    """On each heartbeat, process running jobs that dispatcherd is no longer tracking.
+
+    Jobs still in active_task_ids are legitimately running — leave them alone.
+    For orphaned jobs (not in active_task_ids):
+    - Dispatched (work_unit_id set, controller_node=this_inst) → adopt_job_async (real-time streaming).
+    - Undispatched or not owned by this controller → reap.
+
+    Cross-controller adoption (jobs from a dead peer controller) is deferred to AAP-89602.
+    """
+    workflow_ctype_id = ContentType.objects.get_for_model(WorkflowJob).id
+    base_q = Q(status='running') & (Q(execution_node=this_inst.hostname) | Q(controller_node=this_inst.hostname))
+    base_q &= ~Q(polymorphic_ctype_id=workflow_ctype_id)
+    if active_task_ids:
+        base_q &= ~Q(celery_task_id__in=active_task_ids)
+    if ref_time:
+        base_q &= Q(started__lte=ref_time)
+    jobs = list(UnifiedJob.objects.filter(base_q))
+    if not jobs:
+        return
+
+    for j in jobs:
+        try:
+            if j.work_unit_id and j.controller_node == this_inst.hostname:
+                obj, _ = adopt_job_async.apply_async(args=[j.id], queue=get_task_queuename())
+                # Record the adoption task UUID so subsequent heartbeats see this job
+                # as active and skip it — preventing redundant re-adoption dispatches.
+                UnifiedJob.objects.filter(pk=j.id).update(celery_task_id=obj['uuid'])
+            else:
+                reaper.reap_job(j, 'failed')
+        except Exception:
+            logger.exception(f'Failed processing job {j.id} in heartbeat job loop')
+
+
+@task(queue=get_task_queuename, timeout=3600 * 2, on_duplicate='discard')
+def adopt_job_async(job_id):
+    """Adopt a single orphaned job in a background task, streaming events in real-time.
+
+    Called from _process_startup_jobs and _process_running_jobs via apply_async so the
+    heartbeat returns immediately. on_duplicate='discard' ensures only one adoption runs
+    per job across heartbeat cycles.
+    """
+    job = UnifiedJob.objects.filter(id=job_id, status='running').first()
+    if not job:
+        logger.debug(f'adopt_job_async: job {job_id} is no longer running, skipping')
+        return
+
+    adoption_timeout = settings.HADR_JOB_ADOPTION_TIMEOUT
+    timeout_cutoff = now() - timedelta(seconds=adoption_timeout)
+    last_event_time = job.get_event_queryset().aggregate(Max('created'))['created__max']
+    orphaned_since = last_event_time or job.started
+    if orphaned_since and orphaned_since < timeout_cutoff:
+        logger.error(f'Job {job.id} (unit={job.work_unit_id}) orphaned for >{adoption_timeout}s, failing')
+        reaper.reap_job(job, 'failed', job_explanation='Job exceeded HADR_JOB_ADOPTION_TIMEOUT during controller restart')
+        return
+
+    receptor_ctl = get_receptor_ctl()
+    try:
+        reattach_to_work_unit(job, receptor_ctl)
+    except Exception:
+        logger.exception(f'adopt_job_async: adoption failed for job {job.id} (unit={job.work_unit_id})')
+    finally:
+        try:
+            receptor_ctl.close()
+        except Exception:
+            pass
+
+
 @task(queue=get_task_queuename, timeout=1800, on_duplicate='queue_one')
 def awx_receptor_workunit_reaper():
     """

awx/main/tests/functional/tasks/test_tasks_jobs.py

@@ -56,3 +56,89 @@ def test_runjob_run_can_accept_waiting_status(jt_linked, mocker):
 
     mock_pre_run.assert_called_once()
     assert status_at_pre_run == 'running'
+
+
+@pytest.mark.django_db
+def test_finalize_job_run_with_real_job_instance(jt_linked, mocker):
+    """Test that _finalize_job_run executes the real finalization path with actual Job instance."""
+    from awx.main.tasks.jobs import _finalize_job_run
+    from awx.main.tasks.callback import RunnerCallback
+
+    job = jt_linked.create_unified_job()
+    job.status = 'running'
+    job.save()
+
+    # Create a real callback
+    callback = RunnerCallback(model=Job)
+    callback.event_ct = 5
+    callback.host_status_counts = {'ok': 1, 'failed': 0}  # Non-None triggers events_processed_hook
+
+    # Mock the hooks that would be called
+    mocker.patch('awx.main.tasks.jobs.events_processed_hook')
+    mocker.patch('awx.main.tasks.jobs.ScheduleTaskManager')
+    mocker.patch('awx.main.tasks.jobs.ScheduleWorkflowManager')
+
+    # Call the real finalization function
+    result = _finalize_job_run(Job, job.id, callback, 'successful')
+
+    # Verify the job was updated with successful status
+    job.refresh_from_db()
+    assert job.status == 'successful'
+    assert result.id == job.id
+
+
+@pytest.mark.django_db
+def test_finalize_job_run_with_blocked_jobs(jt_linked, mocker):
+    """Test that _finalize_job_run calls ScheduleTaskManager for jobs with blocked dependencies."""
+    from awx.main.tasks.jobs import _finalize_job_run
+    from awx.main.tasks.callback import RunnerCallback
+
+    job = jt_linked.create_unified_job()
+    job.status = 'running'
+    job.save()
+
+    # Create another job that depends on this one
+    blocked_job = jt_linked.create_unified_job()
+    blocked_job.status = 'blocked'
+    blocked_job.save()
+    job.unifiedjob_blocked_jobs.add(blocked_job)
+
+    callback = RunnerCallback(model=Job)
+    callback.event_ct = 0
+
+    # Mock the hooks
+    mock_schedule_task = mocker.patch('awx.main.tasks.jobs.ScheduleTaskManager')
+    mocker.patch('awx.main.tasks.jobs.events_processed_hook')
+    mocker.patch('awx.main.tasks.jobs.ScheduleWorkflowManager')
+
+    # Call finalization
+    _finalize_job_run(Job, job.id, callback, 'successful')
+
+    # Verify ScheduleTaskManager was called
+    mock_schedule_task.return_value.schedule.assert_called_once()
+
+
+@pytest.mark.django_db
+def test_finalize_job_run_calls_websocket_emit(jt_linked, mocker):
+    """Test that websocket status is emitted after finalization."""
+    from awx.main.tasks.jobs import _finalize_job_run
+    from awx.main.tasks.callback import RunnerCallback
+
+    job = jt_linked.create_unified_job()
+    job.status = 'running'
+    job.save()
+
+    callback = RunnerCallback(model=Job)
+    callback.event_ct = 0
+
+    # Mock websocket_emit_status
+    mocker.patch('awx.main.tasks.jobs.events_processed_hook')
+    mocker.patch('awx.main.tasks.jobs.ScheduleTaskManager')
+    mocker.patch('awx.main.tasks.jobs.ScheduleWorkflowManager')
+    mock_websocket = mocker.patch.object(type(job), 'websocket_emit_status')
+
+    # Call finalization with failed status
+    _finalize_job_run(Job, job.id, callback, 'failed')
+
+    # Verify websocket was called with correct status
+    mock_websocket.assert_called_with('failed')

awx/main/tests/functional/tasks/test_tasks_system.py

@@ -6,6 +6,7 @@ import tempfile
 import shutil
 from contextlib import contextmanager
 from unittest import mock
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -20,12 +21,16 @@ from awx.main.tasks.system import (
     _reap_and_mark_lost_instance,
     inspect_execution_and_hop_nodes,
     _heartbeat_instance_management,
+    _process_startup_jobs,
+    _process_running_jobs,
+    _startup_reap_undispatched,
 )
+from awx.main.dispatch.reaper import reap
 from awx.main.management.commands.dispatcherd import Command
 from django.db import DatabaseError
 from django.utils.timezone import now, timedelta
 
-from awx.main.models import Instance, Inventory, Job, Organization, ReceptorAddress, InstanceLink
+from awx.main.models import Instance, Inventory, Job, Organization, ReceptorAddress, InstanceLink, WorkflowJob
 from awx.main.models.inventory import Group, Host
 
 
@@ -395,8 +400,9 @@ class TestReapAndMarkLostInstance:
     def test_reap_exception_does_not_prevent_mark_offline(self, settings):
         settings.AWX_AUTO_DEPROVISION_INSTANCES = False
         inst = self._inst()
+        Job.objects.create(controller_node=inst.hostname, status='running', work_unit_id=None)
         with mock.patch('awx.main.tasks.system.reaper') as mock_reaper:
-            mock_reaper.reap.side_effect = Exception('receptor timeout')
+            mock_reaper.reap_job.side_effect = Exception('receptor timeout')
             _reap_and_mark_lost_instance(inst)
         inst.refresh_from_db()
         assert inst.node_state == Instance.States.UNAVAILABLE
@@ -543,7 +549,7 @@ def test_heartbeat_defers_lost_instances_when_mesh_gate_blocks(settings):
         mock.patch('awx.main.tasks.system.inspect_execution_and_hop_nodes'),
         mock.patch.object(Instance, 'local_health_check'),
     ):
-        _, _, lost_result = _heartbeat_instance_management()
+        _, _, lost_result, _ = _heartbeat_instance_management()
 
     assert lost_result == []
 
@@ -595,7 +601,7 @@ def test_kubernetes_passes_mesh_gate_with_empty_routing(settings):
         mock.patch('awx.main.tasks.system.inspect_execution_and_hop_nodes'),
         mock.patch.object(Instance, 'local_health_check'),
     ):
-        _, _, lost_result = _heartbeat_instance_management()
+        _, _, lost_result, _ = _heartbeat_instance_management()
 
     # K8s bypasses the gate — lost instances are reaped even with empty routing
     assert len(lost_result) == 1
@@ -637,7 +643,7 @@ def test_mesh_gate_defers_control_but_cleans_execution_hop(settings):
         mock.patch('awx.main.tasks.system.inspect_execution_and_hop_nodes'),
         mock.patch.object(Instance, 'local_health_check'),
     ):
-        _, _, lost_result = _heartbeat_instance_management()
+        _, _, lost_result, _ = _heartbeat_instance_management()
 
     # Execution and hop nodes should be in lost_result (reaped immediately)
     # Control node should be deferred (not in lost_result)
@@ -649,7 +655,7 @@ def test_mesh_gate_defers_control_but_cleans_execution_hop(settings):
 
 @pytest.mark.django_db
 def test_heartbeat_marks_offline_when_receptor_unavailable(settings):
-    """FileNotFoundError from get_receptor_ctl → this_inst marked offline, returns (None, None, None)."""
+    """FileNotFoundError from get_receptor_ctl → this_inst marked offline, returns (None, None, None, None)."""
     settings.CLUSTER_HOST_ID = 'ctrl-0'
     settings.AWX_AUTO_DEPROVISION_INSTANCES = False
 
@@ -663,6 +669,719 @@ def test_heartbeat_marks_offline_when_receptor_unavailable(settings):
     ):
         result = _heartbeat_instance_management()
 
-    assert result == (None, None, None)
+    assert result == (None, None, None, None)
     this_inst.refresh_from_db()
     assert this_inst.node_state == Instance.States.UNAVAILABLE
+
+
+# ── AAP-89607: unified startup job processing ─────────────────────────────────
+
+
+@pytest.mark.django_db
+def test_process_startup_jobs_skips_dispatched_job(me_inst, settings):
+    """_process_startup_jobs() must not reap jobs that have a work_unit_id (dispatched to receptor).
+
+    Dispatched jobs are handed to adopt_job_async for background streaming adoption.
+    """
+    dispatched = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='abc12345')
+    with patch('awx.main.tasks.system.adopt_job_async') as mock_adopt:
+        mock_adopt.apply_async = MagicMock()
+        _process_startup_jobs(me_inst)
+    dispatched.refresh_from_db()
+    assert dispatched.status == 'running', 'dispatched job was wrongly reaped by _process_startup_jobs()'
+    mock_adopt.apply_async.assert_called_once()
+
+
+@pytest.mark.django_db
+def test_process_startup_jobs_reaps_undispatched_job(me_inst, settings):
+    """_process_startup_jobs() reaps jobs with no work_unit_id (never dispatched to receptor)."""
+    undispatched = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=None)
+    _process_startup_jobs(me_inst)
+    undispatched.refresh_from_db()
+    assert undispatched.status == 'failed', 'undispatched job should have been reaped'
+
+
+@pytest.mark.django_db
+def test_reap_reaps_dispatched_jobs(me_inst):
+    """reap() reaps dispatched jobs (no undispatched_only filter)."""
+    dispatched = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-xyz')
+    reap(instance=me_inst)
+    dispatched.refresh_from_db()
+    assert dispatched.status == 'failed', 'reap() should reap dispatched jobs'
+
+
+# ── AAP-89607: adoption loop ──────────────────────────────────────────────────
+
+
+@pytest.mark.django_db
+def test_startup_no_op_when_no_jobs(me_inst):
+    """_process_startup_jobs() is a no-op when there are no running jobs."""
+    with patch('awx.main.tasks.system.adopt_job_async') as mock_adopt:
+        mock_adopt.apply_async = MagicMock()
+        _process_startup_jobs(me_inst)
+    mock_adopt.apply_async.assert_not_called()
+
+
+@pytest.mark.django_db
+def test_adoption_skips_still_running_work_unit(me_inst):
+    """Startup dispatches adopt_job_async for a still-running job — no inline blocking."""
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='running-unit')
+
+    with patch('awx.main.tasks.system.adopt_job_async') as mock_adopt:
+        mock_adopt.apply_async = MagicMock()
+        _process_startup_jobs(me_inst)
+
+    mock_adopt.apply_async.assert_called_once_with(args=[job.id], queue=mock_adopt.apply_async.call_args[1]['queue'])
+    job.refresh_from_db()
+    assert job.status == 'running', 'job must not be reaped — adoption deferred to background task'
+
+
+@pytest.mark.django_db
+def test_adoption_timeout_fails_job(me_inst, settings):
+    """Jobs orphaned longer than HADR_JOB_ADOPTION_TIMEOUT are reaped by adopt_job_async.
+
+    Timeout is measured from the last event received, not from job.started.
+    """
+    settings.HADR_JOB_ADOPTION_TIMEOUT = 3600
+    from awx.main.tasks.system import adopt_job_async
+
+    job = Job.objects.create(
+        controller_node=me_inst.hostname,
+        status='running',
+        work_unit_id='old-unit',
+        started=now() - timedelta(seconds=7200),
+    )
+    with patch('awx.main.tasks.system.reattach_to_work_unit') as mock_reattach:
+        adopt_job_async(job.id)
+    mock_reattach.assert_not_called()
+    job.refresh_from_db()
+    assert job.status == 'failed', 'timed-out job should be reaped by adopt_job_async'
+    assert 'HADR_JOB_ADOPTION_TIMEOUT' in job.job_explanation
+
+
+@pytest.mark.django_db
+def test_adoption_timeout_spares_long_running_job_with_recent_events(me_inst, settings):
+    """Long-running jobs are NOT killed if events arrived recently.
+
+    A job started 2 hours ago but with events 5 minutes ago (brief outage) must not be failed.
+    Timeout is measured from last_event.created, not job.started.
+    """
+    from awx.main.models import JobEvent
+    from awx.main.tasks.system import adopt_job_async
+
+    settings.HADR_JOB_ADOPTION_TIMEOUT = 3600
+    job = Job.objects.create(
+        controller_node=me_inst.hostname,
+        status='running',
+        work_unit_id='long-running-unit',
+        started=now() - timedelta(seconds=7200),
+    )
+    JobEvent.objects.create(job=job, counter=10, event='runner_on_ok', job_created=job.created)
+    JobEvent.objects.filter(job=job).update(created=now() - timedelta(seconds=300))
+
+    with patch('awx.main.tasks.system.get_receptor_ctl'), patch('awx.main.tasks.system.reattach_to_work_unit') as mock_reattach:
+        adopt_job_async(job.id)
+    mock_reattach.assert_called_once()
+    job.refresh_from_db()
+    assert job.status == 'running', 'Long-running job with recent events must not be failed'
+
+
+@pytest.mark.django_db
+def test_adoption_finalizes_successful_job(me_inst):
+    """reattach_to_work_unit finalizes a job as successful when ExitCode=0."""
+    from awx.main.tasks.receptor import reattach_to_work_unit
+
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='done-unit')
+    ctl = MagicMock()
+    ctl.simple_command.return_value = {'StateName': 'Succeeded', 'ExitCode': 0}
+
+    with (
+        patch('awx.main.tasks.receptor.AWXReceptorJob') as mock_job_cls,
+        patch('awx.main.tasks.callback.RunnerCallback'),
+    ):
+        mock_instance = MagicMock()
+        mock_instance._process_phase.return_value = MagicMock(status='successful', rc=0)
+        mock_job_cls.return_value = mock_instance
+
+        reattach_to_work_unit(job, ctl)
+
+    job.refresh_from_db()
+    assert job.status == 'successful'
+
+
+@pytest.mark.django_db
+def test_adoption_finalizes_failed_job(me_inst):
+    """reattach_to_work_unit finalizes a job as failed when ExitCode=1."""
+    from awx.main.tasks.receptor import reattach_to_work_unit
+
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='failed-unit')
+    ctl = MagicMock()
+    ctl.simple_command.return_value = {'StateName': 'Failed', 'ExitCode': 1}
+
+    with (
+        patch('awx.main.tasks.receptor.AWXReceptorJob') as mock_job_cls,
+        patch('awx.main.tasks.callback.RunnerCallback'),
+    ):
+        mock_instance = MagicMock()
+        mock_instance._process_phase.return_value = MagicMock(status='failed', rc=1)
+        mock_job_cls.return_value = mock_instance
+
+        reattach_to_work_unit(job, ctl)
+
+    job.refresh_from_db()
+    assert job.status == 'failed'
+
+
+@pytest.mark.django_db
+def test_process_running_jobs_adopts_dispatched_skips_active(me_inst):
+    """_process_running_jobs() dispatches adopt_job_async for orphaned jobs, skips active ones."""
+    active_job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='active-unit', celery_task_id='active-uuid')
+    orphaned_dispatched = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='orphaned-unit', celery_task_id='orphan-uuid')
+
+    dispatched_ids = []
+    with patch('awx.main.tasks.system.adopt_job_async') as mock_adopt:
+        mock_adopt.apply_async = MagicMock(side_effect=lambda args, **kw: dispatched_ids.append(args[0]))
+        _process_running_jobs(me_inst, active_task_ids={'active-uuid'}, ref_time=None)
+
+    assert orphaned_dispatched.id in dispatched_ids, 'orphaned dispatched job should be queued for adoption'
+    assert active_job.id not in dispatched_ids, 'active job should not be touched'
+
+
+@pytest.mark.django_db
+def test_process_running_jobs_reaps_undispatched(me_inst):
+    """_process_running_jobs() reaps undispatched orphaned jobs."""
+    undispatched = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=None, celery_task_id='orphan-undispatched')
+    _process_running_jobs(me_inst, active_task_ids=set(), ref_time=None)
+    undispatched.refresh_from_db()
+    assert undispatched.status == 'failed', 'undispatched orphaned job should be reaped'
+
+
+@pytest.mark.django_db
+def test_process_running_jobs_noop_when_no_orphaned_jobs(me_inst):
+    """_process_running_jobs() is a no-op (early return) when all running jobs are active."""
+    active_job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='u1', celery_task_id='active-uuid')
+    with patch('awx.main.tasks.system.adopt_job_async') as mock_adopt:
+        mock_adopt.apply_async = MagicMock()
+        _process_running_jobs(me_inst, active_task_ids={'active-uuid'}, ref_time=None)
+    mock_adopt.apply_async.assert_not_called()
+    active_job.refresh_from_db()
+    assert active_job.status == 'running'
+
+
+@pytest.mark.django_db
+def test_process_running_jobs_ref_time_filter(me_inst):
+    """_process_running_jobs() excludes jobs started after ref_time."""
+    future_job = Job.objects.create(
+        controller_node=me_inst.hostname,
+        status='running',
+        work_unit_id=None,
+        celery_task_id='future-uuid',
+        started=now() + timedelta(seconds=60),
+    )
+    ref_time = now()
+    with patch('awx.main.tasks.system.adopt_job_async') as mock_adopt:
+        mock_adopt.apply_async = MagicMock()
+        _process_running_jobs(me_inst, active_task_ids=set(), ref_time=ref_time)
+    mock_adopt.apply_async.assert_not_called()
+    future_job.refresh_from_db()
+    assert future_job.status == 'running', 'job started after ref_time should not be touched'
+
+
+@pytest.mark.django_db
+def test_process_running_jobs_exception_does_not_abort_loop(me_inst):
+    """Exception in _process_running_jobs job loop is caught — remaining jobs still processed."""
+    job1 = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-a', celery_task_id='uuid-a')
+    job2 = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-b', celery_task_id='uuid-b')
+
+    dispatched = []
+
+    def side_effect(args, **kw):
+        if args[0] == job1.id:
+            raise RuntimeError('simulated failure')
+        dispatched.append(args[0])
+
+    with patch('awx.main.tasks.system.adopt_job_async') as mock_adopt:
+        mock_adopt.apply_async = MagicMock(side_effect=side_effect)
+        _process_running_jobs(me_inst, active_task_ids=set(), ref_time=None)
+
+    assert job2.id in dispatched, 'second job must be processed even though first raised'
+
+
+@pytest.mark.django_db
+def test_process_startup_jobs_exception_does_not_abort_loop(me_inst, settings):
+    """Exception in _process_startup_jobs job loop is caught — remaining jobs still processed."""
+    settings.HADR_JOB_ADOPTION_TIMEOUT = 3600
+    job1 = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-err')
+    job2 = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=None)
+
+    def side_effect(args, **kw):
+        if args[0] == job1.id:
+            raise RuntimeError('simulated failure')
+
+    with patch('awx.main.tasks.system.adopt_job_async') as mock_adopt:
+        mock_adopt.apply_async = MagicMock(side_effect=side_effect)
+        _process_startup_jobs(me_inst)
+
+    job2.refresh_from_db()
+    assert job2.status == 'failed', 'undispatched job2 must still be reaped after job1 exception'
+
+
+@pytest.mark.django_db
+def test_startup_reap_undispatched_reaps_undispatched_leaves_dispatched(me_inst):
+    """_startup_reap_undispatched() reaps jobs with no work_unit_id, leaves dispatched jobs alone.
+
+    This is the unconditional safety net called from _run_dispatch_startup_common for cases
+    where cluster_node_heartbeat returns early (receptor unavailable, rejoining cluster).
+    """
+    undispatched = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=None)
+    dispatched = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='abc-unit')
+    _startup_reap_undispatched(me_inst.hostname)
+    undispatched.refresh_from_db()
+    dispatched.refresh_from_db()
+    assert undispatched.status == 'failed', 'undispatched job should be reaped by safety net'
+    assert dispatched.status == 'running', 'dispatched job must not be reaped (adoption loop handles it)'
+
+
+@pytest.mark.django_db
+def test_startup_reap_undispatched_no_op_when_no_jobs(me_inst):
+    """_startup_reap_undispatched() is a no-op when no undispatched running jobs exist."""
+    _startup_reap_undispatched(me_inst.hostname)  # should not raise
+
+
+@pytest.mark.django_db
+def test_startup_reap_undispatched_skips_workflow_jobs(me_inst):
+    """_startup_reap_undispatched() must not reap WorkflowJobs.
+
+    WorkflowJob has work_unit_id=None (it never owns a receptor work unit) but must not
+    be reaped on controller restart — it coordinates via its node jobs, not receptor directly.
+    """
+    wfj = WorkflowJob.objects.create(status='running', controller_node=me_inst.hostname)
+    _startup_reap_undispatched(me_inst.hostname)
+    wfj.refresh_from_db()
+    assert wfj.status == 'running', 'WorkflowJob must not be reaped by startup safety net'
+
+
+@pytest.mark.django_db
+def test_process_startup_jobs_skips_workflow_jobs(me_inst, settings):
+    """_process_startup_jobs() must not reap WorkflowJobs on controller restart.
+
+    WorkflowJob has work_unit_id=None but is not a receptor-dispatched job; reaping it
+    on startup would immediately fail running workflows.
+    """
+    settings.HADR_JOB_ADOPTION_TIMEOUT = 3600
+    wfj = WorkflowJob.objects.create(status='running', controller_node=me_inst.hostname)
+    _process_startup_jobs(me_inst)
+    wfj.refresh_from_db()
+    assert wfj.status == 'running', 'WorkflowJob must not be reaped by startup job loop'
+
+
+@pytest.mark.django_db
+def test_configure_runner_callback_populates_host_map(me_inst):
+    """RunnerCallback.configure_for_job populates host_map from inventory hosts for adoption."""
+    from awx.main.tasks.callback import RunnerCallback
+    from awx.main.models import Organization, Inventory, Host
+
+    org = Organization.objects.first()
+    inv = Inventory.objects.create(name='cb-hostmap-inv', organization=org)
+    host = Host.objects.create(name='myhost', inventory=inv)
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-hm', inventory=inv)
+
+    cb = RunnerCallback(model=Job)
+    cb.configure_for_job(job)
+
+    assert cb.host_map.get('myhost') == host.id
+
+
+@pytest.mark.django_db
+def test_compute_adoption_dedup_no_events(me_inst):
+    """_compute_adoption_dedup returns (0, set()) when job has no events in DB."""
+    from awx.main.tasks.receptor import _compute_adoption_dedup
+
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-1')
+    threshold, collision_zone = _compute_adoption_dedup(job)
+    assert threshold == 0
+    assert collision_zone == set()
+
+
+@pytest.mark.django_db
+def test_compute_adoption_dedup_contiguous_events(me_inst):
+    """_compute_adoption_dedup returns (max_counter, empty set) for contiguous events."""
+    from awx.main.tasks.receptor import _compute_adoption_dedup
+    from awx.main.models import JobEvent
+
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-1')
+    for ctr in [1, 2, 3, 4, 5]:
+        JobEvent.objects.create(job=job, counter=ctr, event='runner_on_ok')
+
+    threshold, collision_zone = _compute_adoption_dedup(job)
+    assert threshold == 5
+    assert collision_zone == set()
+
+
+@pytest.mark.django_db
+def test_compute_adoption_dedup_gap_produces_collision_zone(me_inst):
+    """_compute_adoption_dedup finds gap and puts above-gap events in collision_zone."""
+    from awx.main.tasks.receptor import _compute_adoption_dedup
+    from awx.main.models import JobEvent
+
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-1')
+    # Events 1-3 contiguous, then gap at 4, then 5 and 6 committed out of order
+    for ctr in [1, 2, 3, 5, 6]:
+        JobEvent.objects.create(job=job, counter=ctr, event='runner_on_ok')
+
+    threshold, collision_zone = _compute_adoption_dedup(job)
+    assert threshold == 3
+    assert collision_zone == {5, 6}
+
+
+@pytest.mark.django_db
+def test_adopt_job_async_exception_is_swallowed(me_inst, settings):
+    """adopt_job_async swallows exceptions from reattach_to_work_unit and does not re-raise."""
+    from awx.main.tasks.system import adopt_job_async
+
+    settings.HADR_JOB_ADOPTION_TIMEOUT = 3600
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-err')
+    with patch('awx.main.tasks.system.get_receptor_ctl'), patch('awx.main.tasks.system.reattach_to_work_unit', side_effect=RuntimeError('network failure')):
+        adopt_job_async(job.id)  # must not raise
+
+
+@pytest.mark.django_db
+def test_adoption_counter_skip_dedup():
+    """RunnerCallback.event_handler hybrid dedup: O(1) threshold + small collision-zone set.
+
+    safe_threshold skips the contiguous prefix with a single integer comparison.
+    persisted_counters (collision_zone) handles out-of-order committed events above
+    the threshold — bounded by worker concurrency, not total event count.
+    """
+    from awx.main.tasks.callback import RunnerCallback
+
+    cb = RunnerCallback()
+    # Simulate: events 1-8 committed contiguously (threshold=8),
+    # events 9,10 committed out-of-order (collision zone)
+    cb.dedup_threshold = 8
+    cb.persisted_counters = {9, 10}
+
+    dispatched = []
+    cb.dispatcher = MagicMock()
+    cb.dispatcher.dispatch.side_effect = dispatched.append
+
+    from collections import deque
+
+    cb.instance = MagicMock()
+    cb.instance.event_class.WRAPUP_EVENT = 'playbook_on_stats'
+    cb.event_data_key = 'job_id'
+    cb.job_created = None
+    cb.parent_workflow_job_id = None
+    cb.host_map = {}
+    cb.recent_event_timings = deque(maxlen=100)
+
+    # below threshold — skipped by O(1) integer check
+    cb.event_handler({'event': 'runner_on_ok', 'counter': 5, 'job_id': 1})
+    assert len(dispatched) == 0, 'counter <= threshold should be skipped'
+
+    # at threshold — also skipped
+    cb.event_handler({'event': 'runner_on_ok', 'counter': 8, 'job_id': 1})
+    assert len(dispatched) == 0, 'counter == threshold should be skipped'
+
+    # in collision zone — skipped by set membership
+    cb.event_handler({'event': 'runner_on_ok', 'counter': 9, 'job_id': 1})
+    assert len(dispatched) == 0, 'counter in collision_zone should be skipped'
+
+    # None threshold disables dedup entirely (normal job path, zero overhead)
+    cb.dedup_threshold = None
+    cb.persisted_counters = None
+    cb.event_handler({'event': 'runner_on_ok', 'counter': 1, 'job_id': 1})
+    # no assertion — just verify no exception
+
+
+# ── reattach_to_work_unit branch coverage ────────────────────────────────────
+
+
+@pytest.mark.django_db
+def test_reattach_receptor_command_fails(me_inst):
+    """receptor_ctl.simple_command raising returns False without touching the job."""
+    from awx.main.tasks.receptor import reattach_to_work_unit
+
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-err')
+    ctl = MagicMock()
+    ctl.simple_command.side_effect = Exception('connection refused')
+
+    result = reattach_to_work_unit(job, ctl)
+
+    assert result is False
+    job.refresh_from_db()
+    assert job.status == 'running'
+
+
+@pytest.mark.django_db
+def test_reattach_exit_code_from_detail(me_inst):
+    """Exit code is parsed from the Detail string when ExitCode is absent."""
+    from awx.main.tasks.receptor import reattach_to_work_unit
+
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-detail')
+    ctl = MagicMock()
+    ctl.simple_command.return_value = {'StateName': 'Failed', 'Detail': 'exit status 2'}
+
+    with (
+        patch('awx.main.tasks.receptor.AWXReceptorJob') as mock_job_cls,
+        patch('awx.main.tasks.callback.RunnerCallback'),
+    ):
+        mock_job_cls.return_value = MagicMock()
+        reattach_to_work_unit(job, ctl)
+
+    job.refresh_from_db()
+    assert job.status == 'failed'
+
+
+@pytest.mark.django_db
+def test_reattach_exit_code_fallback_succeeded(me_inst):
+    """_process_phase returning status='successful' → job finalized as successful."""
+    from awx.main.tasks.receptor import reattach_to_work_unit
+
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-fb-ok')
+    ctl = MagicMock()
+    ctl.simple_command.return_value = {'StateName': 'Succeeded', 'Detail': 'not-a-number'}
+
+    with (
+        patch('awx.main.tasks.receptor.AWXReceptorJob') as mock_job_cls,
+        patch('awx.main.tasks.callback.RunnerCallback'),
+    ):
+        mock_instance = MagicMock()
+        mock_instance._process_phase.return_value = MagicMock(status='successful', rc=0)
+        mock_job_cls.return_value = mock_instance
+        reattach_to_work_unit(job, ctl)
+
+    job.refresh_from_db()
+    assert job.status == 'successful'
+
+
+@pytest.mark.django_db
+def test_reattach_exit_code_fallback_failed(me_inst):
+    """Unparseable Detail with Failed state → exit_code=1 → failed."""
+    from awx.main.tasks.receptor import reattach_to_work_unit
+
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-fb-fail')
+    ctl = MagicMock()
+    ctl.simple_command.return_value = {'StateName': 'Failed', 'Detail': ''}
+
+    with (
+        patch('awx.main.tasks.receptor.AWXReceptorJob') as mock_job_cls,
+        patch('awx.main.tasks.callback.RunnerCallback'),
+    ):
+        mock_job_cls.return_value = MagicMock()
+        reattach_to_work_unit(job, ctl)
+
+    job.refresh_from_db()
+    assert job.status == 'failed'
+
+
+@pytest.mark.django_db
+def test_reattach_process_phase_raises(me_inst):
+    """When _process_phase raises, the job is still finalized via the pre-fetched exit_code."""
+    from awx.main.tasks.receptor import reattach_to_work_unit
+
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-raise')
+    ctl = MagicMock()
+    ctl.simple_command.return_value = {'StateName': 'Succeeded', 'ExitCode': 0}
+
+    with (
+        patch('awx.main.tasks.receptor.AWXReceptorJob') as mock_job_cls,
+        patch('awx.main.tasks.callback.RunnerCallback'),
+    ):
+        mock_instance = MagicMock()
+        mock_instance._process_phase.side_effect = RuntimeError('boom')
+        mock_job_cls.return_value = mock_instance
+        result = reattach_to_work_unit(job, ctl)
+
+    assert result is True
+    job.refresh_from_db()
+    assert job.status == 'successful'
+
+
+@pytest.mark.django_db
+def test_reattach_job_already_finalized(me_inst):
+    """finished_callback finalizing the job during process phase is not overwritten."""
+    from awx.main.tasks.receptor import reattach_to_work_unit
+
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-already')
+    ctl = MagicMock()
+    ctl.simple_command.return_value = {'StateName': 'Succeeded', 'ExitCode': 0}
+
+    def _finalize_in_db(*args, **kwargs):
+        Job.objects.filter(pk=job.pk).update(status='successful')
+
+    with (
+        patch('awx.main.tasks.receptor.AWXReceptorJob') as mock_job_cls,
+        patch('awx.main.tasks.callback.RunnerCallback'),
+    ):
+        mock_instance = MagicMock()
+        mock_instance._process_phase.side_effect = _finalize_in_db
+        mock_job_cls.return_value = mock_instance
+        reattach_to_work_unit(job, ctl)
+
+    job.refresh_from_db()
+    assert job.status == 'successful'
+
+
+@pytest.mark.django_db
+def test_adopt_job_async_calls_reattach(me_inst, settings):
+    """adopt_job_async creates its own receptor_ctl and calls reattach_to_work_unit."""
+    from awx.main.tasks.system import adopt_job_async
+
+    settings.HADR_JOB_ADOPTION_TIMEOUT = 3600
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-async')
+
+    with patch('awx.main.tasks.system.get_receptor_ctl') as mock_ctl_factory, patch('awx.main.tasks.system.reattach_to_work_unit') as mock_reattach:
+        adopt_job_async(job.id)
+
+    mock_ctl_factory.assert_called_once()
+    mock_reattach.assert_called_once_with(job, mock_ctl_factory.return_value)
+
+
+@pytest.mark.django_db
+def test_adopt_job_async_skips_already_finalized(me_inst, settings):
+    """adopt_job_async is a no-op if the job is no longer running."""
+    from awx.main.tasks.system import adopt_job_async
+
+    settings.HADR_JOB_ADOPTION_TIMEOUT = 3600
+    job = Job.objects.create(controller_node=me_inst.hostname, status='successful', work_unit_id='unit-done')
+
+    with patch('awx.main.tasks.system.reattach_to_work_unit') as mock_reattach:
+        adopt_job_async(job.id)
+
+    mock_reattach.assert_not_called()
+
+
+@pytest.mark.django_db
+def test_adopt_job_async_reaps_on_timeout(me_inst, settings):
+    """adopt_job_async reaps a job orphaned longer than HADR_JOB_ADOPTION_TIMEOUT."""
+    from awx.main.tasks.system import adopt_job_async
+
+    settings.HADR_JOB_ADOPTION_TIMEOUT = 3600
+    job = Job.objects.create(
+        controller_node=me_inst.hostname,
+        status='running',
+        work_unit_id='unit-old',
+        started=now() - timedelta(seconds=7200),
+    )
+
+    with patch('awx.main.tasks.system.reattach_to_work_unit') as mock_reattach:
+        adopt_job_async(job.id)
+
+    mock_reattach.assert_not_called()
+    job.refresh_from_db()
+    assert job.status == 'failed'
+
+
+@pytest.mark.django_db
+def test_adopt_job_async_ctl_close_exception_is_swallowed(me_inst, settings):
+    """adopt_job_async does not propagate exceptions from receptor_ctl.close()."""
+    from awx.main.tasks.system import adopt_job_async
+
+    settings.HADR_JOB_ADOPTION_TIMEOUT = 3600
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-close-err')
+
+    mock_ctl = MagicMock()
+    mock_ctl.close.side_effect = RuntimeError('socket already closed')
+
+    with patch('awx.main.tasks.system.get_receptor_ctl', return_value=mock_ctl), patch('awx.main.tasks.system.reattach_to_work_unit'):
+        adopt_job_async(job.id)  # must not raise
+
+
+# ── _finalize_job_run coverage ──────────────────────────────────────────────
+
+
+@pytest.mark.django_db
+def test_finalize_job_run_with_extra_fields(me_inst):
+    """_finalize_job_run includes extra_fields in the update."""
+    from awx.main.tasks.jobs import _finalize_job_run
+    from awx.main.tasks.callback import RunnerCallback
+
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running')
+    callback = RunnerCallback(model=Job)
+    callback.instance = job
+    callback.wrapup_event_dispatched = True
+
+    _finalize_job_run(Job, job.pk, callback, 'successful', extra_fields={'elapsed': 100.5})
+
+    job.refresh_from_db()
+    assert job.status == 'successful'
+    assert job.elapsed == 100.5
+
+
+@pytest.mark.django_db
+def test_finalize_job_run_update_model_returns_none(me_inst):
+    """_finalize_job_run returns None when update_model fails."""
+    from awx.main.tasks.jobs import _finalize_job_run
+    from awx.main.tasks.callback import RunnerCallback
+
+    callback = RunnerCallback(model=Job)
+    callback.wrapup_event_dispatched = True
+
+    with patch('awx.main.tasks.jobs.update_model', return_value=None):
+        result = _finalize_job_run(Job, 99999, callback, 'successful')
+
+    assert result is None
+
+
+@pytest.mark.django_db
+def test_finalize_job_run_with_blocked_jobs(me_inst):
+    """_finalize_job_run schedules task manager when job has blocked dependents."""
+    from awx.main.tasks.jobs import _finalize_job_run
+    from awx.main.tasks.callback import RunnerCallback
+
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running')
+    callback = RunnerCallback(model=Job)
+    callback.instance = job
+    callback.wrapup_event_dispatched = True
+
+    with patch('awx.main.tasks.jobs.ScheduleTaskManager') as mock_tm:
+        with patch('awx.main.tasks.jobs.update_model') as mock_update:
+            # Create a mock instance that has unifiedjob_blocked_jobs
+            mock_instance = MagicMock()
+            mock_instance.unifiedjob_blocked_jobs.exists.return_value = True
+            mock_update.return_value = mock_instance
+            _finalize_job_run(Job, job.pk, callback, 'successful')
+            mock_tm.return_value.schedule.assert_called()
+
+
+@pytest.mark.django_db
+def test_finalize_job_run_with_inventory_update(me_inst):
+    """_finalize_job_run schedules inventory computed fields update for jobs with inventory."""
+    from awx.main.tasks.jobs import _finalize_job_run
+    from awx.main.tasks.callback import RunnerCallback
+    from awx.main.models import Organization, Inventory
+
+    org = Organization.objects.first()
+    inv = Inventory.objects.create(name='test-inv', organization=org)
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', inventory=inv)
+
+    callback = RunnerCallback(model=Job)
+    callback.instance = job
+    callback.wrapup_event_dispatched = True
+
+    with patch('awx.main.tasks.jobs.update_inventory_computed_fields') as mock_delay:
+        _finalize_job_run(Job, job.pk, callback, 'successful')
+        mock_delay.delay.assert_called_once_with(inv.id)
+
+
+@pytest.mark.django_db
+def test_finalize_job_run_inventory_update_exception_logged(me_inst):
+    """_finalize_job_run logs exception from inventory computed fields update."""
+    from awx.main.tasks.jobs import _finalize_job_run
+    from awx.main.tasks.callback import RunnerCallback
+    from awx.main.models import Organization, Inventory
+
+    org = Organization.objects.first()
+    inv = Inventory.objects.create(name='test-inv-err', organization=org)
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', inventory=inv)
+
+    callback = RunnerCallback(model=Job)
+    callback.instance = job
+    callback.wrapup_event_dispatched = True
+
+    with patch('awx.main.tasks.jobs.update_inventory_computed_fields') as mock_delay:
+        mock_delay.delay.side_effect = RuntimeError('celery error')
+        with patch('awx.main.tasks.jobs.logger') as mock_logger:
+            _finalize_job_run(Job, job.pk, callback, 'successful')
+            mock_logger.exception.assert_called()

awx/main/tests/functional/test_jobs.py

@@ -199,7 +199,6 @@ class TestAnsibleFactsSave:
 
 @pytest.mark.django_db
 def test_update_hosts_resolved_deadlock(inventory, mocker):
-
     hosts = [Host.objects.create(inventory=inventory, name=f'foo{i}') for i in range(3)]
 
     # Set ansible_facts for each host

awx/main/tests/unit/tasks/test_jobs.py

@@ -661,3 +661,209 @@ class TestRunInventoryUpdatePopulateWorkloadIdentityTokens:
 
         # The instance's get_cloud_credential should now return the same object with context
         assert task.instance.get_cloud_credential() is cloud_cred
+
+    def test_no_cloud_credential_when_none_returned_from_instance(self):
+        """When instance.get_cloud_credential returns None, populate_workload_identity_tokens is called with None."""
+        task = jobs.RunInventoryUpdate()
+        task.instance = mock.MagicMock()
+        task.instance.get_cloud_credential.return_value = None
+        task._credentials = []
+
+        with mock.patch.object(jobs.BaseTask, 'populate_workload_identity_tokens') as mock_super:
+            task.populate_workload_identity_tokens()
+
+        mock_super.assert_called_once_with(additional_credentials=None)
+
+
+def test_work_unit_released_on_quota_exceeded():
+    """
+    Work unit is released on quota exceeded (when res is None).
+
+    Scenario: Job hits quota limit → _handle_work_error returns None →
+    BaseTask.run() early-returns. Work unit MUST be released.
+    """
+    receptor_ctl = mock.MagicMock()
+    receptor_job = mock.MagicMock()
+    receptor_job.unit_id = "test-unit-123"
+    receptor_job.receptor_ctl = receptor_ctl
+
+    # When res is None (quota exceeded case)
+    res = None
+
+    # This is the fixed code path from jobs.py lines 810-818
+    if not res:
+        if receptor_job and getattr(receptor_job, 'receptor_ctl', None):
+            try:
+                receptor_job._receptor_release_work(receptor_job.receptor_ctl, 'error')
+            except Exception:
+                pass
+
+    # Verify release was called
+    receptor_job._receptor_release_work.assert_called_once_with(receptor_ctl, 'error')
+
+
+def test_work_unit_not_released_if_no_receptor_ctl():
+    """
+    Edge case: receptor_ctl not set (transmit phase failed before unit submission).
+
+    Verify graceful handling when receptor_ctl is None.
+    """
+    receptor_job = mock.MagicMock(spec=['unit_id', '_receptor_release_work'])
+    receptor_job.receptor_ctl = None  # Not set
+
+    res = None
+
+    # Code should not crash
+    if not res:
+        if receptor_job and getattr(receptor_job, 'receptor_ctl', None):
+            receptor_job._receptor_release_work(receptor_job.receptor_ctl, 'error')
+        # else: No release needed, unit was never submitted
+
+    # Verify _receptor_release_work was NOT called
+    assert receptor_job._receptor_release_work.call_count == 0
+
+
+@mock.patch('awx.main.tasks.jobs.ScheduleWorkflowManager')
+@mock.patch('awx.main.tasks.jobs.ScheduleTaskManager')
+@mock.patch('awx.main.tasks.jobs.events_processed_hook')
+@mock.patch('awx.main.tasks.jobs.update_model')
+def test_finalize_job_run_with_host_status_counts(mock_update, mock_hook, mock_task_mgr, mock_workflow_mgr):
+    """events_processed_hook is called when host_status_counts is not None."""
+    instance = mock.MagicMock()
+    instance.host_status_counts = {'successful': 5}  # Not None
+    instance.unifiedjob_blocked_jobs.exists.return_value = False
+    instance.spawned_by_workflow = False
+    instance.inventory_id = None
+    mock_update.return_value = instance
+
+    cb = mock.MagicMock()
+    cb.get_delayed_update_fields.return_value = {}
+    cb.wrapup_event_dispatched = True
+
+    jobs._finalize_job_run(mock.MagicMock, pk=1, runner_callback=cb, status='successful')
+
+    mock_hook.assert_called_once_with(instance)
+
+
+@mock.patch('awx.main.tasks.jobs.ScheduleWorkflowManager')
+@mock.patch('awx.main.tasks.jobs.ScheduleTaskManager')
+@mock.patch('awx.main.tasks.jobs.events_processed_hook')
+@mock.patch('awx.main.tasks.jobs.update_model')
+def test_finalize_job_run_with_wrapup_not_dispatched(mock_update, mock_hook, mock_task_mgr, mock_workflow_mgr):
+    """events_processed_hook is called when wrapup_event_dispatched is False."""
+    instance = mock.MagicMock()
+    instance.host_status_counts = None
+    instance.unifiedjob_blocked_jobs.exists.return_value = False
+    instance.spawned_by_workflow = False
+    instance.inventory_id = None
+    mock_update.return_value = instance
+
+    cb = mock.MagicMock()
+    cb.get_delayed_update_fields.return_value = {}
+    cb.wrapup_event_dispatched = False  # Not dispatched
+
+    jobs._finalize_job_run(mock.MagicMock, pk=1, runner_callback=cb, status='successful')
+
+    mock_hook.assert_called_once_with(instance)
+
+
+@mock.patch('awx.main.tasks.jobs.ScheduleWorkflowManager')
+@mock.patch('awx.main.tasks.jobs.ScheduleTaskManager')
+@mock.patch('awx.main.tasks.jobs.events_processed_hook')
+@mock.patch('awx.main.tasks.jobs.update_model')
+def test_finalize_job_run_calls_log_lifecycle(mock_update, mock_hook, mock_task_mgr, mock_workflow_mgr):
+    """instance.log_lifecycle('finalize_run') is called."""
+    instance = mock.MagicMock()
+    instance.host_status_counts = None
+    instance.unifiedjob_blocked_jobs.exists.return_value = False
+    instance.spawned_by_workflow = False
+    instance.inventory_id = None
+    mock_update.return_value = instance
+
+    cb = mock.MagicMock()
+    cb.get_delayed_update_fields.return_value = {}
+    cb.wrapup_event_dispatched = True
+
+    jobs._finalize_job_run(mock.MagicMock, pk=1, runner_callback=cb, status='successful')
+
+    instance.log_lifecycle.assert_called_once_with('finalize_run')
+
+
+@mock.patch('awx.main.tasks.jobs.ScheduleWorkflowManager')
+@mock.patch('awx.main.tasks.jobs.ScheduleTaskManager')
+@mock.patch('awx.main.tasks.jobs.events_processed_hook')
+@mock.patch('awx.main.tasks.jobs.update_model')
+def test_finalize_job_run_calls_websocket_emit(mock_update, mock_hook, mock_task_mgr, mock_workflow_mgr):
+    """instance.websocket_emit_status is called with the final status."""
+    instance = mock.MagicMock()
+    instance.host_status_counts = None
+    instance.unifiedjob_blocked_jobs.exists.return_value = False
+    instance.spawned_by_workflow = False
+    instance.inventory_id = None
+    mock_update.return_value = instance
+
+    cb = mock.MagicMock()
+    cb.get_delayed_update_fields.return_value = {}
+    cb.wrapup_event_dispatched = True
+
+    jobs._finalize_job_run(mock.MagicMock, pk=1, runner_callback=cb, status='failed')
+
+    instance.websocket_emit_status.assert_called_once_with('failed')
+
+
+@mock.patch('awx.main.tasks.jobs.fcntl.lockf')
+@mock.patch('awx.main.tasks.jobs.os.open')
+@mock.patch('awx.main.tasks.jobs.os.path.exists')
+def test_acquire_lock_shared(mock_exists, mock_os_open, mock_lockf):
+    """acquire_lock with exclusive=False uses LOCK_SH."""
+    import fcntl
+    from awx.main.tasks.jobs import BaseTask
+
+    # Setup mocks
+    mock_exists.return_value = True
+    mock_os_open.return_value = 5
+    task = BaseTask()
+    task.instance = mock.MagicMock()
+    task.instance.refresh_from_db = mock.MagicMock()
+    task.instance.cancel_flag = False
+
+    project = mock.MagicMock()
+    project.get_lock_file.return_value = '/tmp/project.lock'
+
+    # Mock signal_callback to return False (not cancelled)
+    with mock.patch('awx.main.tasks.jobs.signal_callback', return_value=False):
+        task.acquire_lock(project, unified_job_id=1, exclusive=False)
+
+    # Verify LOCK_SH (not LOCK_EX) was used
+    call_args = mock_lockf.call_args
+    lock_flags = call_args[0][1]
+    assert lock_flags & fcntl.LOCK_SH  # Should have LOCK_SH bit
+
+
+@mock.patch('awx.main.tasks.jobs.fcntl.lockf')
+@mock.patch('awx.main.tasks.jobs.os.open')
+@mock.patch('awx.main.tasks.jobs.os.path.exists')
+def test_acquire_lock_exclusive(mock_exists, mock_os_open, mock_lockf):
+    """acquire_lock with exclusive=True uses LOCK_EX."""
+    import fcntl
+    from awx.main.tasks.jobs import BaseTask
+
+    # Setup mocks
+    mock_exists.return_value = True
+    mock_os_open.return_value = 5
+    task = BaseTask()
+    task.instance = mock.MagicMock()
+    task.instance.refresh_from_db = mock.MagicMock()
+    task.instance.cancel_flag = False
+
+    project = mock.MagicMock()
+    project.get_lock_file.return_value = '/tmp/project.lock'
+
+    # Mock signal_callback to return False (not cancelled)
+    with mock.patch('awx.main.tasks.jobs.signal_callback', return_value=False):
+        task.acquire_lock(project, unified_job_id=1, exclusive=True)
+
+    # Verify LOCK_EX was used
+    call_args = mock_lockf.call_args
+    lock_flags = call_args[0][1]
+    assert lock_flags & fcntl.LOCK_EX  # Should have LOCK_EX bit

awx/main/tests/unit/test_tasks.py

@@ -469,11 +469,12 @@ class TestGenericRun:
         task.model.objects.get = mock.Mock(return_value=job)
         task.build_private_data_files = mock.Mock(side_effect=OSError())
 
+        with mock.patch('awx.main.tasks.jobs.update_model', return_value=job) as mock_update_model:
             with mock.patch('awx.main.tasks.jobs.shutil.copytree'), mock.patch('awx.main.tasks.jobs.evaluate_policy'):
                 with pytest.raises(Exception):
                     task.run(1)
 
-        update_model_call = task.update_model.call_args[1]
+        update_model_call = mock_update_model.call_args[1]
         assert 'OSError' in update_model_call['result_traceback']
         assert update_model_call['status'] == 'error'
         assert update_model_call['emitted_events'] == 0
@@ -574,11 +575,12 @@ class TestAdhocRun(TestJobExecution):
         task.model.objects.get = mock.Mock(return_value=adhoc_job)
         task.build_inventory = mock.Mock()
 
+        with mock.patch('awx.main.tasks.jobs.update_model', return_value=adhoc_job) as mock_update_model:
             with pytest.raises(Exception):
                 task.run(adhoc_job.pk)
 
         call_args, _ = task.update_model.call_args_list[0]
-        update_model_call = task.update_model.call_args[1]
+        update_model_call = mock_update_model.call_args[1]
         assert 'Jinja variables are not allowed' in update_model_call['result_traceback']
 
     '''

awx/settings/defaults.py

@@ -223,6 +223,9 @@ JOB_EVENT_WORKERS = 4
 # Minimum number of workers for the dispatcher (dispatcherd) process pool
 DISPATCHER_MIN_WORKERS = 4
 
+# Maximum number of events buffered per callback worker before flushing via bulk_create()
+JOB_EVENT_CALLBACK_BUFFER_SIZE = 1000
+
 # The number of seconds to buffer callback receiver bulk
 # writes in memory before flushing via JobEvent.objects.bulk_create()
 JOB_EVENT_BUFFER_SECONDS = 1
@@ -418,6 +421,12 @@ CLUSTER_NODE_HEARTBEAT_PERIOD = 60
 CLUSTER_NODE_MISSED_HEARTBEAT_TOLERANCE = 2
 
 RECEPTOR_SERVICE_ADVERTISEMENT_PERIOD = 60  # https://github.com/ansible/receptor/blob/aa1d589e154d8a0cb99a220aff8f98faf2273be6/pkg/netceptor/netceptor.go#L34
+
+# Seconds after a controller restart before an orphaned dispatched job is failed.
+# Jobs with work_unit_id set are skipped by the reaper and handled by the adoption loop
+# (_attempt_adoption_for_dispatched_jobs). This timeout is the maximum time we wait for
+# the EE to finish before giving up and failing the job.
+HADR_JOB_ADOPTION_TIMEOUT = 3600
 EXECUTION_NODE_REMEDIATION_CHECKS = 60 * 30  # once every 30 minutes check if an execution node errors have been resolved
 
 # Amount of time dispatcher will try to reconnect to database for jobs and consuming new work