Cover the remaining heartbeat decision branches

gate_lost_instances now takes the mesh check as a callable and skips it
when nothing is lost, so the heartbeat has no branch of its own there.
New tests cover a lost instance's job the decision doesn't reap, and a
heartbeat job the decision skips.

Co-Authored-By: Claude <noreply@anthropic.com>

awx/main/tasks/adoption_decisions.py

@@ -44,8 +44,12 @@ def gate_lost_instances(lost_instances, mesh_ready):
     While the receptor mesh is re-establishing, a control node may only look lost, so its
     handling waits for a later heartbeat. Execution and hop nodes don't depend on mesh
     consensus and are handled either way.
+
+    mesh_ready is a callable, asked only when something is lost.
     """
-    if mesh_ready:
+    if not lost_instances:
+        return []
+    if mesh_ready():
         return list(lost_instances)
     return [inst for inst in lost_instances if inst.node_type in ('execution', 'hop')]
 

awx/main/tasks/system.py

@@ -828,8 +828,7 @@ def _heartbeat_instance_management():
             logger.error("Cluster Host Not Found: {}".format(settings.CLUSTER_HOST_ID))
             return None, None, None, None
 
-    if lost_instances:
-        lost_instances = gate_lost_instances(lost_instances, _mesh_all_ready_nodes_visible(mesh_status))
+    lost_instances = gate_lost_instances(lost_instances, lambda: _mesh_all_ready_nodes_visible(mesh_status))
 
     return this_inst, instance_list, lost_instances, ctl
 

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

@@ -10,6 +10,7 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
+from awx.main.tasks.adoption_decisions import JobAction
 from awx.main.tasks.system import (
     CleanupImagesAndFiles,
     execution_node_health_check,
@@ -1385,3 +1386,35 @@ def test_finalize_job_run_inventory_update_exception_logged(me_inst):
         with patch('awx.main.tasks.jobs.logger') as mock_logger:
             _finalize_job_run(Job, job.pk, callback, 'successful')
             mock_logger.exception.assert_called()
+
+
+@pytest.mark.django_db
+def test_lost_instance_job_not_marked_for_reaping_is_left_running(settings):
+    """Only jobs the decision says to reap are reaped; AAP-89602 will return ADOPT here."""
+    settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+    lost = Instance.objects.create(hostname='lost-node', node_type='control', node_state='ready')
+    job = Job.objects.create(controller_node='lost-node', status='running', work_unit_id='unit-1')
+
+    with patch('awx.main.tasks.system.lost_instance_job_action', return_value=JobAction.ADOPT), patch('awx.main.tasks.system.reaper') as mock_reaper:
+        _reap_and_mark_lost_instance(lost)
+
+    mock_reaper.reap_job.assert_not_called()
+    job.refresh_from_db()
+    assert job.status == 'running'
+
+
+@pytest.mark.django_db
+def test_heartbeat_job_the_decision_skips_is_neither_adopted_nor_reaped(me_inst):
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-1', started=now() - timedelta(minutes=1))
+
+    with (
+        patch('awx.main.tasks.system.running_job_action', return_value=JobAction.SKIP),
+        patch('awx.main.tasks.system.adopt_job_async') as mock_adopt,
+        patch('awx.main.tasks.system.reaper') as mock_reaper,
+    ):
+        _process_running_jobs(me_inst, ['other-task'], now())
+
+    mock_adopt.apply_async.assert_not_called()
+    mock_reaper.reap_job.assert_not_called()
+    job.refresh_from_db()
+    assert job.status == 'running'

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

@@ -99,14 +99,20 @@ class TestFindLostInstances:
 class TestGateLostInstances:
     def test_mesh_ready_handles_every_lost_instance(self):
         lost = [make_instance(PEER), make_instance(EXEC, node_type='execution'), make_instance('hop-1', node_type='hop')]
-        assert gate_lost_instances(lost, mesh_ready=True) == lost
+        assert gate_lost_instances(lost, mesh_ready=lambda: True) == lost
 
     def test_mesh_not_ready_defers_only_control_nodes(self):
         control = make_instance(PEER)
         execution = make_instance(EXEC, node_type='execution')
         hop = make_instance('hop-1', node_type='hop')
         hybrid = make_instance('hybrid-1', node_type='hybrid')
-        assert gate_lost_instances([control, execution, hop, hybrid], mesh_ready=False) == [execution, hop]
+        assert gate_lost_instances([control, execution, hop, hybrid], mesh_ready=lambda: False) == [execution, hop]
+
+    def test_mesh_is_not_checked_when_nothing_is_lost(self):
+        def mesh_ready():
+            raise AssertionError('mesh checked with nothing lost')
+
+        assert gate_lost_instances([], mesh_ready) == []
 
 
 class TestLostInstanceDecisions: