ansible · hsong-rh · Sep 22, 2026
diff --git a/awx/main/tasks/callback.py b/awx/main/tasks/callback.py
@@ -131,10 +131,13 @@ def configure_for_job(self, instance, safe_env=None, dedup_threshold=None, persi
         Having both paths call this method means any missing field is caught by the
         normal test suite rather than only by adoption-specific tests.
 
+        host_map is deliberately not set here — each path has a different cheapest source
+        for it. See populate_host_map.
+
         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)
+            safe_env: masked environment values that win over the pattern masking applied in
+                status_handler. None → {}; status_handler still pattern-masks in that case.
             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).
@@ -143,7 +146,7 @@ def configure_for_job(self, instance, safe_env=None, dedup_threshold=None, persi
                 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.job_created = str(instance.created)  # stamped on every event by event_handler
         self.safe_env = safe_env if safe_env is not None else {}
         if getattr(instance, 'spawned_by_workflow', False):
             try:
@@ -152,20 +155,6 @@ def configure_for_job(self, instance, safe_env=None, dedup_threshold=None, persi
                 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):
@@ -174,6 +163,35 @@ def create_for_job(cls, instance, safe_env=None, dedup_threshold=None, persisted
         callback.configure_for_job(instance, safe_env, dedup_threshold, persisted_counters)
         return callback
 
+    @staticmethod
+    def inventory_script_params(instance):
+        """Arguments for Inventory.get_script_data, matching what build_inventory writes."""
+        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
+        return script_params
+
+    def populate_host_map(self, script_data):
+        """Map hostname -> remote_tower_id so event_handler can stamp host_id on events."""
+        for hostname, hv in script_data.get('_meta', {}).get('hostvars', {}).items():
+            self.host_map[hostname] = hv.get('remote_tower_id', '')
+
+    def populate_host_map_from_inventory(self, instance):
+        """Adoption path: fetch script_data ourselves, since no inventory file is written.
+
+        The normal path hands us the script_data build_inventory already built, so it never
+        pays for this query. Sourcing host_map from the same script_data either way is what
+        keeps adoption honoring inventory type (smart, constructed), enabled state and
+        job slicing.
+        """
+        if not (hasattr(instance, 'inventory') and instance.inventory_id):
+            return
+        try:
+            self.populate_host_map(instance.inventory.get_script_data(**self.inventory_script_params(instance)))
+        except Exception:
+            pass  # host_map stays {}; host_id won't be set on replayed events
+
     def event_handler(self, event_data):
         #
         # ⚠️  D-D-D-DANGER ZONE ⚠️
@@ -310,9 +328,18 @@ def status_handler(self, status_data, runner_config):
         Ansible runner callback triggered on status transition
         """
         if status_data['status'] == 'starting':
-            job_env = dict(runner_config.env)
+            from awx.main.models.credential import build_safe_env  # Circular import
+
+            '''
+            Pattern-mask first so nothing sensitive can reach job_env (exposed on the job
+            detail API) even when safe_env is empty — which is the case during adoption of a
+            job whose original controller died before its 'starting' status was persisted.
+            In the normal path safe_env is already build_safe_env(env), so this is a no-op.
+            '''
+            job_env = build_safe_env(runner_config.env)
             '''
-            Take the safe environment variables and overwrite
+            Take the safe environment variables and overwrite. These win over the pattern
+            masking above: they also cover credential-plugin values whose names match no pattern.
             '''
             for k, v in self.safe_env.items():
                 if k in job_env:

diff --git a/awx/main/tasks/jobs.py b/awx/main/tasks/jobs.py
@@ -521,15 +521,14 @@ def build_env(self, instance, private_data_dir, private_data_files=None):
 
     def write_inventory_file(self, inventory, private_data_dir, file_name, script_params):
         script_data = inventory.get_script_data(**script_params)
+        # Reuse the script_data we are about to write rather than making the callback fetch
+        # its own copy, which is what the adoption path has to do.
+        self.runner_callback.populate_host_map(script_data)
         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 = {"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_params = self.runner_callback.inventory_script_params(instance)
         return self.write_inventory_file(instance.inventory, private_data_dir, 'hosts', script_params)
 
     def build_args(self, instance, private_data_dir, passwords):

diff --git a/awx/main/tasks/receptor.py b/awx/main/tasks/receptor.py
@@ -888,7 +888,16 @@ def _get_adoption_exit_code(unit_status, state_name):
 
 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)
+    callback = RunnerCallback.create_for_job(
+        job,
+        safe_env=dict(job.job_env or {}),
+        dedup_threshold=dedup_threshold,
+        persisted_counters=collision_zone,
+    )
+    # Normal runs get host_map for free from build_inventory. Adoption writes no inventory
+    # file, so it has to source the same data itself or replayed events lose host_id.
+    callback.populate_host_map_from_inventory(job)
+    return callback
 
 
 def _get_or_create_private_data_dir(job):
@@ -928,44 +937,100 @@ def _finalize_adopted_job(job, callback, exit_code, process_phase_failed):
     if job.started:
         extra['elapsed'] = (finished_at - job.started).total_seconds()
 
+    # Record adoption metadata in job_explanation (must be before finalization). Both paths
+    # through this function are adoptions, and which controller took the job over matters
+    # most when the process phase raised, so this is unconditional.
+    # Goes through delay_update rather than extra_fields: _finalize_job_run applies
+    # extra_fields on top of the delayed fields, so setting it here would discard any
+    # explanation status_handler recorded for the real failure. delay_update appends.
+    # settings.CLUSTER_HOST_ID rather than Instance.objects.me().hostname: me() looks the row
+    # up *by* CLUSTER_HOST_ID, so the two are the same string, and me() additionally raises
+    # when no row matches. Letting that escape would skip finalization while the caller's
+    # `finally` still releases the work unit, stranding the job in `running` forever.
+    surviving_controller = settings.CLUSTER_HOST_ID
+    callback.delay_update(job_explanation=f'Job adopted by {surviving_controller}. Work unit: {job.work_unit_id}. Execution node: {job.execution_node}.')
+
     _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.
+    """Return (safe_threshold, collision_zone, persisted_ct) 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.
+        prefix starting from counter 1). 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.
+
+    persisted_ct: how many events are already in the DB, counted in the database and
+        including any beyond the cap. Seeds callback.event_ct, which would otherwise
+        undercount precisely when the collision zone is truncated.
     """
-    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
+    qs = job.get_event_queryset()
+
+    # The contiguous prefix has to be anchored at counter 1 — if the first event never
+    # committed, every later counter sits after a gap and none of them can be skipped.
+    # Two constant-cost queries: one anchor check, one "lowest counter whose successor is
+    # missing". Walking the counters in windows instead would cost a query per window and
+    # fetch every row, on every adoption attempt.
+    if not qs.filter(counter=1).exists():
+        safe_threshold = 0
+    else:
+        next_ctr = qs.filter(counter=OuterRef('counter') + 1)
+        gap_event = qs.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))
+    above_threshold = qs.filter(counter__gt=safe_threshold)
+    # count() in the database rather than len(list(...)): an early gap leaves nearly every
+    # event above the threshold, and materializing that list is the OOM this cap exists to
+    # prevent. Slicing the queryset lets the DB apply the limit too.
+    persisted_above = above_threshold.count()
 
-    if len(collision_zone_list) > cap:
+    if persisted_above > 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'Job {job.id}: collision_zone has {persisted_above} events above safe_threshold, '
+            f'exceeds dedup cap of {cap}. Events beyond the 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
+    # order_by keeps truncation deterministic and retains the counters closest to the
+    # contiguous prefix; slicing an unordered queryset would drop arbitrary rows.
+    collision_zone = set(above_threshold.order_by('counter').values_list('counter', flat=True)[:cap])
+    return safe_threshold, collision_zone, safe_threshold + persisted_above
 
 
-def reattach_to_work_unit(job, receptor_ctl):
+def get_adoption_unit_status(receptor_ctl, job):
+    """Return the receptor work unit status dict for an adoption attempt.
+
+    Asks this controller's own receptor first. The unit is local whenever this controller
+    submitted the work (same-controller restart, where execution_node is a remote EE but
+    the proxy unit lives here) or already adopted it on an earlier heartbeat. Only the
+    local query reports a real StateName — `work adopt` answers 'Already Adopted' with no
+    state, which would disable the Pending/Running check in reattach_to_work_unit.
+
+    Falls back to adopting from the execution node when the local receptor does not know
+    the unit: the genuine cross-controller case, and the case where this controller's unit
+    directory was lost (e.g. /tmp wiped on an OCP pod restart).
+    """
+    unit_id = job.work_unit_id
+    try:
+        return receptor_ctl.simple_command(f'work status {unit_id}')
+    except Exception:
+        if not job.execution_node or job.execution_node == settings.CLUSTER_HOST_ID:
+            raise
+        logger.info(f'Job {job.id}: unit {unit_id} unknown to local receptor, adopting from execution node {job.execution_node}')
+        return adopt_remote_work(receptor_ctl, job.execution_node, unit_id)
+
+
+def reattach_to_work_unit(job, receptor_ctl, unit_status=None):
     """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
@@ -974,33 +1039,38 @@ def reattach_to_work_unit(job, receptor_ctl):
     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.
+    blocked. Supports cross-controller adoption via job.execution_node (AAP-89602).
+
+    Args:
+        unit_status: Optional pre-fetched work unit status dict (avoids duplicate adopt_remote_work calls)
     """
     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}')
+        if unit_status is None:
+            unit_status = get_adoption_unit_status(receptor_ctl, job)
+        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')
     except Exception:
-        logger.warning(f'Cannot get receptor status for work unit {unit_id} (job {job.id}), deferring adoption')
+        logger.warning(f'Cannot get receptor status for work unit {unit_id} (job {job.id}, execution_node={job.execution_node}), 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'):
+    if state_name in RECEPTOR_ACTIVE_STATES:
         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)
+    safe_threshold, collision_zone, persisted_ct = _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)
+    # Account for events already persisted so emitted_events / EOF final_counter reflect the
+    # full job. Uses the DB count rather than len(collision_zone), which is capped.
+    callback.event_ct = persisted_ct
 
     private_data_dir = _get_or_create_private_data_dir(job)
     adoption_task = _AdoptionTask(job, callback)