Reproduce the job adoption failure scenarios as functional tests

The failpoint harness found 29 defects on devel by breaking a two-controller
docker-compose cluster. 22 of them are now reproduced here with the four
seams, in seconds and without a cluster:

  test_scenarios_liveness.py     a controller judged lost while alive
  test_scenarios_adoption.py     finalize errors, the quiet-job deadline
  test_scenarios_job_control.py  cancels of orphaned jobs, reaped jobs whose
                                 playbook keeps running
  test_scenarios_data.py         a reaped job's stats, host ids on replay,
                                 activity stream, files left on disk

Each defect test asserts the behavior the job should have and is marked
xfail(strict=True) with its Jira bug. When a fix lands, the test passes,
strict xfail fails the run, and the marker comes off.

The fast-clock test in test_adoption_clock.py now asserts the expected
behavior the same way. The decision unit tests that describe the inputs
behind three of these bugs point at them.

Regression tests cover what devel already gets right: repeated restarts,
a 20,000 event replay, a lost controller's waiting jobs, and a controller
gone from the mesh having its job reaped.

The other 7 defects need real process restarts, Redis, the scheduler's
clock or timing under load, and stay with the harness.

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

awx/main/tests/adoption_cluster.py

@@ -0,0 +1,92 @@
+"""A small cluster in the test database, for job adoption failure scenarios.
+
+The scenarios were first found with the failpoint harness on a two-controller
+docker-compose cluster. Here the same cluster is rows in the test database:
+
+  ctrl-0      this controller, running the heartbeat under test
+  ctrl-1      a peer controller, judged lost when its last_seen is stale
+  receptor-1  the execution node the jobs run on
+
+Receptor is replaced with FakeReceptorWork and the callback-receiver queue with
+RecordingDispatcher (see awx.main.tests.fake_receptor), so the real heartbeat, reap,
+adoption and finalize code runs without a mesh.
+"""
+
+import contextlib
+from unittest import mock
+
+from django.utils.timezone import now, timedelta
+
+from awx.main.models import Instance
+from awx.main.tasks.system import _heartbeat_handle_lost_instances, _heartbeat_instance_management
+from awx.main.tests.fake_receptor import RecordingDispatcher
+
+THIS = 'ctrl-0'
+PEER = 'ctrl-1'
+EXEC = 'receptor-1'
+UNIT = 'unit-1'
+GRACE = 120  # CLUSTER_NODE_HEARTBEAT_PERIOD * CLUSTER_NODE_MISSED_HEARTBEAT_TOLERANCE
+
+
+def configure(settings):
+    settings.CLUSTER_HOST_ID = THIS
+    settings.AWX_AUTO_DEPROVISION_INSTANCES = False
+    settings.CLUSTER_NODE_HEARTBEAT_PERIOD = 60
+    settings.CLUSTER_NODE_MISSED_HEARTBEAT_TOLERANCE = 2
+
+
+def make_cluster(peer_seconds_since_seen=GRACE + 80):
+    """Create ctrl-0 (healthy), ctrl-1 (silent for peer_seconds_since_seen) and receptor-1."""
+    t = now()
+    this = Instance.objects.create(hostname=THIS, node_type='control', node_state='ready', last_seen=t)
+    peer = Instance.objects.create(hostname=PEER, node_type='control', node_state='ready', last_seen=t - timedelta(seconds=peer_seconds_since_seen))
+    Instance.objects.create(hostname=EXEC, node_type='execution', node_state='ready', last_seen=t)
+    return this, peer
+
+
+def mesh(*connected):
+    """Receptor status as ctrl-0 sees it, with ctrl-0 connected to each node in connected."""
+    return {'KnownConnectionCosts': {THIS: {node: 1 for node in connected}}, 'Advertisements': []}
+
+
+def run_lost_instance_handling(mesh_status, work=None, clock=None):
+    """Run ctrl-0's heartbeat up to and including lost-instance handling. Return the lost hostnames.
+
+    work, when given, is returned wherever system.py opens ReceptorWork, so a test can see
+    whether a work unit was canceled.
+    """
+    ctl = mock.MagicMock()
+    ctl.simple_command.return_value = mesh_status
+    with contextlib.ExitStack() as stack:
+        stack.enter_context(mock.patch('awx.main.tasks.system.get_receptor_ctl', return_value=ctl))
+        stack.enter_context(mock.patch('awx.main.tasks.system.inspect_execution_and_hop_nodes'))
+        stack.enter_context(mock.patch.object(Instance, 'local_health_check'))
+        if work is not None:
+            stack.enter_context(mock.patch('awx.main.tasks.system.ReceptorWork', return_value=work))
+        this_inst, _, lost, _ = _heartbeat_instance_management(clock)
+        _heartbeat_handle_lost_instances(lost, this_inst)
+    return sorted(inst.hostname for inst in lost)
+
+
+@contextlib.contextmanager
+def adoption_env(settings, tmp_path):
+    """Patch what adoption touches outside the database: Redis, websockets and the schedulers."""
+    settings.AWX_ISOLATION_BASE_PATH = str(tmp_path)
+    RecordingDispatcher.reset()
+    # connections.close_all() would end the test's transaction; events go to the recorder instead of Redis.
+    with (
+        mock.patch('awx.main.tasks.receptor.connections'),
+        mock.patch('awx.main.tasks.callback.CallbackQueueDispatcher', RecordingDispatcher),
+        mock.patch('awx.main.tasks.jobs.ScheduleTaskManager'),
+        mock.patch('awx.main.tasks.jobs.ScheduleWorkflowManager'),
+        mock.patch('awx.main.models.unified_jobs.UnifiedJob.websocket_emit_status'),
+    ):
+        yield
+
+
+def assert_playbook_not_left_running(job, work):
+    """Fail when the job is no longer active but its work unit was never canceled."""
+    job.refresh_from_db()
+    still_active = job.status in ('running', 'waiting')
+    unit_canceled = ('cancel', job.work_unit_id) in work.calls
+    assert still_active or unit_canceled, f'job is {job.status!r} but its work unit {job.work_unit_id} was never canceled'

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

@@ -63,15 +63,17 @@ def test_peer_is_lost_only_after_the_grace_period(heartbeat_settings):
     assert lost_hostnames(clock) == ['ctrl-1']
 
 
+@pytest.mark.xfail(strict=True, reason='AAP-96827: a controller with a fast clock marks a healthy peer lost')
 @pytest.mark.django_db
-def test_judge_with_a_fast_clock_finds_a_healthy_peer_lost(heartbeat_settings):
-    # ctrl-0's clock runs five minutes fast, so its own last_seen is five minutes ahead.
-    # ctrl-1 heartbeated just now by the correct clock and is healthy.
+def test_judge_with_a_fast_clock_does_not_find_a_healthy_peer_lost(heartbeat_settings):
+    # 2.8 Clock skew. ctrl-0's clock runs five minutes fast, so its own last_seen is five
+    # minutes ahead. ctrl-1 heartbeated just now by the correct clock and is healthy.
+    # Today ctrl-0 judges it by its own clock and finds it lost.
     t = now()
     control_node('ctrl-0', t + timedelta(minutes=5))
     control_node('ctrl-1', t)
 
-    assert lost_hostnames(FakeClock(t + timedelta(minutes=5))) == ['ctrl-1']
+    assert lost_hostnames(FakeClock(t + timedelta(minutes=5))) == []
 
 
 @pytest.mark.django_db

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

@@ -0,0 +1,118 @@
+"""Failure scenarios in the adoption lifecycle: claim, stream, finalize (write-up section 3).
+
+Tests marked xfail reproduce a defect found on devel and assert the behavior it should
+have. The others guard behavior devel already gets right.
+"""
+
+from unittest import mock
+
+import pytest
+from django.db import DatabaseError
+from django.utils.timezone import now, timedelta
+
+from awx.main.models import Job, JobEvent
+from awx.main.tasks import jobs
+from awx.main.tasks.system import adopt_job
+from awx.main.tests.adoption_cluster import UNIT, adoption_env
+from awx.main.tests.fake_clock import FakeClock
+from awx.main.tests.fake_receptor import FakeReceptorWork, FakeWorkUnit, RecordingDispatcher, make_events
+
+TIMEOUT = 300  # HADR_JOB_ADOPTION_TIMEOUT
+
+
+@pytest.fixture(autouse=True)
+def env(settings, tmp_path):
+    settings.HADR_JOB_ADOPTION_TIMEOUT = TIMEOUT
+    with adoption_env(settings, tmp_path):
+        yield
+
+
+@pytest.fixture
+def orphan(me_inst):
+    """A running job this controller lost track of, started a minute ago."""
+    return Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=UNIT, started=now() - timedelta(seconds=60))
+
+
+def save_events(job, counters, created=None):
+    for counter in counters:
+        JobEvent.objects.create(job=job, counter=counter, event='runner_on_ok', job_created=job.created)
+    if created:
+        JobEvent.objects.filter(job=job).update(created=created)
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96828: an error while finalizing a resumed job releases its unit and strands the job')
+@pytest.mark.django_db
+def test_finalize_error_is_retried_on_the_next_adoption(orphan):
+    # 3.5 Finalize strand. The first adoption streams the whole unit, then finalization
+    # raises. Today the unit is released anyway, so the next adoption finds nothing and the
+    # job stays running until the adoption timeout fails it.
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(make_events(10))})
+    real_finalize = jobs._finalize_job_run
+    attempts = []
+
+    def finalize_fails_once(*args, **kwargs):
+        attempts.append(1)
+        if len(attempts) == 1:
+            raise DatabaseError('connection lost during finalize')
+        return real_finalize(*args, **kwargs)
+
+    with mock.patch('awx.main.tasks.jobs._finalize_job_run', side_effect=finalize_fails_once):
+        adopt_job(orphan.id, open_work=lambda: work)
+        adopt_job(orphan.id, open_work=lambda: work)
+
+    orphan.refresh_from_db()
+    assert orphan.status == 'successful'
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96829: a restart fails a healthy job with a long quiet task while its playbook keeps running')
+@pytest.mark.django_db
+def test_quiet_job_whose_unit_is_still_running_is_not_failed(orphan):
+    # 3.6 Quiet deadline. The playbook is in a long task with no output, so its last event
+    # is older than HADR_JOB_ADOPTION_TIMEOUT, but receptor reports the unit running.
+    # Today adoption measures the deadline from the last event and fails the job, while
+    # the playbook keeps running on the execution node.
+    clock = FakeClock()
+    save_events(orphan, range(1, 6), created=clock() - timedelta(seconds=TIMEOUT + 60))
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(make_events(10), state='Running')})
+
+    adopt_job(orphan.id, open_work=lambda: work, clock=clock)
+
+    orphan.refresh_from_db()
+    assert orphan.status == 'running'
+
+
+@pytest.mark.django_db
+def test_adoption_waits_through_repeated_restarts_and_replays_each_event_once(orphan):
+    # 3.7 Three restarts during one job. While the unit runs, each adoption defers; once
+    # it finishes, the replay sends only the events not yet saved, each once.
+    clock = FakeClock()
+    save_events(orphan, range(1, 13), created=clock())
+    unit = FakeWorkUnit(make_events(30), state='Running')
+    work = FakeReceptorWork({UNIT: unit})
+
+    for _ in range(3):
+        adopt_job(orphan.id, open_work=lambda: work, clock=clock)
+        orphan.refresh_from_db()
+        assert orphan.status == 'running'
+        assert RecordingDispatcher.dispatched == []
+        clock.advance(60)
+
+    unit.state = 'Succeeded'
+    adopt_job(orphan.id, open_work=lambda: work, clock=clock)
+
+    orphan.refresh_from_db()
+    assert orphan.status == 'successful'
+    assert RecordingDispatcher.counters() == list(range(13, 31))
+
+
+@pytest.mark.django_db
+def test_large_replay_sends_every_event_once(orphan):
+    # 5.8 Large replay. A 20,000 event unit with the first 500 already saved.
+    save_events(orphan, range(1, 501))
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(make_events(20000))})
+
+    adopt_job(orphan.id, open_work=lambda: work)
+
+    orphan.refresh_from_db()
+    assert orphan.status == 'successful'
+    assert RecordingDispatcher.counters() == list(range(501, 20001))

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

@@ -0,0 +1,139 @@
+"""Failure scenarios about the data a job leaves behind after adoption or a reap (write-up section 5).
+
+Tests marked xfail reproduce a defect found on devel and assert the behavior it should have.
+"""
+
+import json
+import os
+from unittest import mock
+
+import pytest
+
+from awx.main.models import ActivityStream, Job
+from awx.main.tasks.system import adopt_job, awx_receptor_workunit_reaper
+from awx.main.tests.adoption_cluster import EXEC, PEER, UNIT, adoption_env, configure, make_cluster, mesh, run_lost_instance_handling
+from awx.main.tests.fake_receptor import FakeReceptorWork, FakeWorkUnit, RecordingDispatcher, make_events
+
+
+@pytest.fixture(autouse=True)
+def env(settings, tmp_path):
+    configure(settings)
+    with adoption_env(settings, tmp_path):
+        yield
+
+
+def xfail(bug, summary):
+    return pytest.mark.xfail(strict=True, reason=f'{bug}: {summary}')
+
+
+# ── A reaped job's own result ─────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    'bug',
+    [
+        pytest.param('AAP-96839', id='AAP-96839-relaunch-on-failed-hosts', marks=xfail('AAP-96839', 'a reaped job cannot be relaunched on its failed hosts')),
+        pytest.param('AAP-96846', id='AAP-96846-host-metrics', marks=xfail('AAP-96846', 'a reaped job records no host metrics')),
+        pytest.param('AAP-96847', id='AAP-96847-host-pointers', marks=xfail('AAP-96847', 'a reaped job leaves hosts showing the previous job')),
+    ],
+)
+@pytest.mark.django_db
+def test_stats_of_a_reaped_job_are_saved_when_its_unit_finishes(me_inst, settings, bug):
+    # 4.20, 5.10, 5.12. The job was marked failed while its playbook ran on, and the
+    # playbook then finished. Its stats event, which writes host summaries, host metrics
+    # and each host's last job, is never read: the work unit reaper cancels and releases
+    # the finished unit. So relaunch on failed hosts has no summaries to use, the hosts
+    # count nothing, and they keep pointing at the previous job.
+    settings.RECEPTOR_RELEASE_WORK = True
+    Job.objects.create(
+        controller_node=me_inst.hostname, execution_node=EXEC, status='failed', work_unit_id=UNIT, job_explanation='Job reaped due to instance shutdown'
+    )
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(make_events(10))})
+
+    with (
+        mock.patch('awx.main.tasks.system.get_receptor_ctl'),
+        mock.patch('awx.main.tasks.system.ReceptorWork', return_value=work),
+        mock.patch('awx.main.tasks.system.administrative_workunit_reaper'),
+    ):
+        awx_receptor_workunit_reaper()
+
+    assert 'playbook_on_stats' in [event.get('event') for event in RecordingDispatcher.dispatched]
+
+
+# ── Host ids on replayed events ───────────────────────────────────────────────
+
+
+def host_events(hostnames):
+    """Runner events for one task on each host, then the stats event."""
+    events = make_events(len(hostnames) + 2)
+    for event, hostname in zip(events[1:-1], hostnames):
+        event['event_data'] = {'host': hostname}
+    return events
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96843: replayed events link to the wrong hosts after an inventory change')
+@pytest.mark.django_db
+def test_replayed_events_keep_the_host_ids_the_job_started_with(me_inst, inventory):
+    # 5.2 Host map after an inventory change. A normal run maps each host name to the id
+    # it had when the job started. Adoption rebuilds the map from the inventory as it is
+    # now, so after a change while the job was orphaned, replayed events lose the deleted
+    # and disabled hosts and point a recreated host at its new id.
+    hosts = {name: inventory.hosts.create(name=name) for name in ('fp-h1', 'fp-h2', 'fp-h3', 'fp-h4', 'fp-h5')}
+    ids_at_start = {name: host.id for name, host in hosts.items()}
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=UNIT, inventory=inventory)
+
+    # While the job is orphaned: delete fp-h2, recreate fp-h3, disable fp-h4.
+    hosts['fp-h2'].delete()
+    hosts['fp-h3'].delete()
+    inventory.hosts.create(name='fp-h3')
+    hosts['fp-h4'].enabled = False
+    hosts['fp-h4'].save()
+
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(host_events(list(hosts)))})
+    adopt_job(job.id, open_work=lambda: work)
+
+    replayed = {event['host_name']: event.get('host_id') for event in RecordingDispatcher.dispatched if event.get('host_name')}
+    assert replayed == ids_at_start
+
+
+# ── What records a reap ───────────────────────────────────────────────────────
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96849: the activity stream does not record that a job was reaped')
+@pytest.mark.django_db
+def test_reap_is_recorded_in_the_activity_stream():
+    # 5.16 Activity stream. Nothing records that a lost controller's job was marked
+    # failed, so an auditor cannot tell a reaped job from one whose playbook failed.
+    make_cluster()
+    job = Job.objects.create(controller_node=PEER, execution_node=EXEC, status='running', work_unit_id=UNIT)
+    entries_before = ActivityStream.objects.filter(job=job).count()
+
+    run_lost_instance_handling(mesh(EXEC))
+
+    job.refresh_from_db()
+    assert job.status == 'failed'
+    assert ActivityStream.objects.filter(job=job).count() > entries_before
+
+
+# ── Files left on disk ────────────────────────────────────────────────────────
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96852: a dead controller\'s job files stay on disk after it returns')
+@pytest.mark.django_db
+def test_resumed_job_removes_its_original_private_data_dir(me_inst, settings):
+    # 5.20 Credentials left on disk. The job's private data directory from before the
+    # restart holds its inventory and extra variables. Adoption works in a new directory
+    # and removes only that one, so the original stays until periodic cleanup's grace
+    # period (60 minutes by default) has passed.
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=UNIT)
+    original = os.path.join(settings.AWX_ISOLATION_BASE_PATH, f'awx_{job.pk}_before_restart')
+    os.makedirs(os.path.join(original, 'env'))
+    with open(os.path.join(original, 'env', 'extravars'), 'w') as f:
+        json.dump({'survey_password': 'secret'}, f)
+    work = FakeReceptorWork({UNIT: FakeWorkUnit(make_events(10))})
+
+    adopt_job(job.id, open_work=lambda: work)
+
+    job.refresh_from_db()
+    assert job.status == 'successful'
+    assert not os.path.exists(original), f'{original} is still on disk'

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

@@ -0,0 +1,177 @@
+"""Failure scenarios about canceling, capacity and launches around orphaned jobs (write-up section 4).
+
+Tests marked xfail reproduce a defect found on devel and assert the behavior it should
+have. The others guard behavior devel already gets right.
+
+Several devel bugs share one cause: when a lost controller's job is marked failed, its
+playbook keeps running on the execution node. Whatever depends on the job being active
+(launch blocking, relaunch, instance group limits, execution node capacity, workflow
+branching) then sees a finished job. test_reaped_job_does_not_leave_its_playbook_running
+covers each of those bugs with the job shape its scenario used.
+"""
+
+import pytest
+
+from awx.main.models import InstanceGroup, Job, JobTemplate, WorkflowJob, WorkflowJobNode
+from awx.main.tasks.system import adopt_job
+from awx.main.tests.adoption_cluster import (
+    EXEC,
+    PEER,
+    UNIT,
+    adoption_env,
+    assert_playbook_not_left_running,
+    configure,
+    make_cluster,
+    mesh,
+    run_lost_instance_handling,
+)
+from awx.main.tests.fake_receptor import FakeReceptorWork, FakeWorkUnit, make_events
+
+
+@pytest.fixture(autouse=True)
+def env(settings, tmp_path):
+    configure(settings)
+    with adoption_env(settings, tmp_path):
+        yield
+
+
+def peer_job(**fields):
+    """A job controlled by ctrl-1 and running on receptor-1."""
+    fields.setdefault('work_unit_id', UNIT)
+    return Job.objects.create(controller_node=PEER, execution_node=EXEC, status='running', celery_task_id='peer-task', **fields)
+
+
+def running_unit():
+    return FakeReceptorWork({UNIT: FakeWorkUnit(make_events(10), state='Running')})
+
+
+# ── Canceling an orphaned job ─────────────────────────────────────────────────
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96830: canceling a job whose controller is down is dropped; the job never ends canceled')
+@pytest.mark.django_db
+def test_cancel_requested_while_orphaned_cancels_the_unit(me_inst):
+    # 4.1 Cancel while orphaned. The user canceled while the controller was down, so the
+    # cancel reached no one. Today adoption never reads cancel_flag: it waits for the unit
+    # and finishes the job, successful, with cancel_flag set.
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=UNIT, cancel_flag=True)
+    work = running_unit()
+
+    adopt_job(job.id, open_work=lambda: work)
+
+    job.refresh_from_db()
+    assert ('cancel', UNIT) in work.calls
+    assert job.status == 'canceled'
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96835: canceling a job while it is finalized after a resume leaves it running forever')
+@pytest.mark.django_db
+def test_canceled_job_whose_unit_is_gone_is_finalized_canceled(me_inst):
+    # 4.14 Cancel during the adopter's finalize. The cancel stopped the adoption after the
+    # unit was released but before the status was saved. Today every later adoption finds
+    # no unit and defers, so the job stays running forever.
+    job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=UNIT, cancel_flag=True)
+
+    adopt_job(job.id, open_work=lambda: FakeReceptorWork())
+
+    job.refresh_from_db()
+    assert job.status == 'canceled'
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96836: canceling many orphaned jobs marks them failed one by one instead of canceled')
+@pytest.mark.django_db
+def test_canceled_orphans_of_a_lost_controller_end_canceled():
+    # 4.15 Canceling many orphans at once. Twenty jobs of a dead controller were canceled
+    # while it was down. Today they are marked failed, "Job reaped due to instance shutdown".
+    make_cluster()
+    canceled = [peer_job(work_unit_id=f'unit-{n}', cancel_flag=True) for n in range(20)]
+
+    run_lost_instance_handling(mesh(EXEC))
+
+    statuses = {job.status for job in Job.objects.filter(pk__in=[j.pk for j in canceled])}
+    assert statuses == {'canceled'}
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96840: canceling a sliced job leaves the orphaned slice failed with its playbook running')
+@pytest.mark.django_db
+def test_canceled_slice_of_a_lost_controller_ends_canceled_and_stops():
+    # 4.21 Canceling a sliced job. The parent was canceled, so the orphaned slice has
+    # cancel_flag set. Today it ends failed and its playbook runs to the end.
+    make_cluster()
+    job = peer_job(job_slice_number=1, job_slice_count=2, cancel_flag=True)
+    work = running_unit()
+
+    run_lost_instance_handling(mesh(EXEC), work=work)
+
+    job.refresh_from_db()
+    assert job.status == 'canceled'
+    assert ('cancel', UNIT) in work.calls
+
+
+# ── A reaped job whose playbook keeps running ─────────────────────────────────
+
+
+def job_of_template_without_simultaneous_runs():
+    return peer_job(job_template=JobTemplate.objects.create(name='same-hosts', allow_simultaneous=False))
+
+
+def job_in_workflow():
+    workflow = WorkflowJob.objects.create(name='outer', status='running')
+    job = peer_job()
+    WorkflowJobNode.objects.create(workflow_job=workflow, job=job)
+    return job
+
+
+def job_in_limited_instance_group():
+    return peer_job(instance_group=InstanceGroup.objects.create(name='limited', max_concurrent_jobs=2))
+
+
+def xfail(bug):
+    return pytest.mark.xfail(strict=True, reason=f'{bug}: a reaped job leaves its playbook running')
+
+
+@pytest.mark.parametrize(
+    'make_job',
+    [
+        pytest.param(job_of_template_without_simultaneous_runs, id='AAP-96832-same-hosts', marks=xfail('AAP-96832')),
+        pytest.param(job_in_workflow, id='AAP-96833-nested-workflow', marks=xfail('AAP-96833')),
+        pytest.param(job_of_template_without_simultaneous_runs, id='AAP-96834-relaunch', marks=xfail('AAP-96834')),
+        pytest.param(job_in_limited_instance_group, id='AAP-96837-instance-group-limit', marks=xfail('AAP-96837')),
+        pytest.param(peer_job, id='AAP-96841-execution-node-capacity', marks=xfail('AAP-96841')),
+        pytest.param(job_in_workflow, id='AAP-96844-workflow-branch', marks=xfail('AAP-96844')),
+        pytest.param(job_in_workflow, id='AAP-96845-set-stats-artifacts', marks=xfail('AAP-96845')),
+    ],
+)
+@pytest.mark.django_db
+def test_reaped_job_does_not_leave_its_playbook_running(make_job):
+    # ctrl-1 is lost while its job runs on receptor-1, which is still up. Today the job is
+    # marked failed and nothing cancels its work unit, so the playbook runs on unobserved:
+    # the next launch overlaps it (4.6, 4.12), it stops counting toward instance group and
+    # execution node limits (4.17, 4.23), and a workflow takes the failure branch for a
+    # playbook that succeeds (4.11, 5.5, 5.6). Either the unit is canceled when the job is
+    # marked failed, or the job stays active until the unit ends.
+    make_cluster()
+    job = make_job()
+    work = running_unit()
+
+    run_lost_instance_handling(mesh(EXEC), work=work)
+
+    assert_playbook_not_left_running(job, work)
+
+
+# ── Waiting jobs of a lost controller ─────────────────────────────────────────
+
+
+@pytest.mark.django_db
+def test_waiting_jobs_of_a_lost_controller_return_to_pending():
+    # 4.5 Jobs waiting when the controller dies. They had not started, so they go back to
+    # the task manager and run once, elsewhere.
+    make_cluster()
+    waiting = Job.objects.create(controller_node=PEER, execution_node=EXEC, status='waiting')
+
+    run_lost_instance_handling(mesh(EXEC))
+
+    waiting.refresh_from_db()
+    assert waiting.status == 'pending'
+    assert waiting.controller_node == ''
+    assert waiting.execution_node == ''

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

@@ -0,0 +1,63 @@
+"""Failure scenarios about when a controller is really gone (write-up section 2).
+
+Tests marked xfail reproduce a defect found on devel and assert the behavior it should
+have. Each names its Jira bug; when the bug is fixed the test passes, strict xfail fails
+the run, and the marker comes off.
+"""
+
+from unittest import mock
+
+import pytest
+
+from awx.main.models import Job
+from awx.main.tests.adoption_cluster import EXEC, PEER, UNIT, configure, make_cluster, mesh, run_lost_instance_handling
+
+
+@pytest.fixture(autouse=True)
+def cluster_settings(settings):
+    configure(settings)
+
+
+@pytest.fixture
+def peer_job():
+    """A job controlled by ctrl-1 and running on receptor-1."""
+    return Job.objects.create(controller_node=PEER, execution_node=EXEC, status='running', work_unit_id=UNIT, celery_task_id='peer-task')
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96824: a controller that misses heartbeats is judged lost while it still runs its job')
+@pytest.mark.django_db
+def test_slow_controller_still_on_the_mesh_keeps_its_job(peer_job):
+    # 2.1 Slow controller. ctrl-1 has missed two heartbeats but is still connected on the
+    # receptor mesh and still streaming its job. Today ctrl-0 judges it lost from last_seen
+    # alone and marks the live job failed.
+    make_cluster()
+
+    run_lost_instance_handling(mesh(PEER, EXEC))
+
+    peer_job.refresh_from_db()
+    assert peer_job.status == 'running'
+
+
+@pytest.mark.django_db
+def test_controller_gone_from_the_mesh_has_its_job_reaped(peer_job):
+    # The case the slow-controller test must not break: a controller that is silent and
+    # gone from the mesh is lost, and its job is no longer reported as running.
+    make_cluster()
+
+    assert run_lost_instance_handling(mesh(EXEC)) == [PEER]
+
+    peer_job.refresh_from_db()
+    assert peer_job.status == 'failed'
+
+
+@pytest.mark.xfail(strict=True, reason='AAP-96848: a slow controller\'s job first notifies failed with no hosts, then successful')
+@pytest.mark.django_db
+def test_slow_controller_job_sends_no_failed_notification(peer_job):
+    # 5.15 Notification content. The reap of a live controller's job sends a failed
+    # notification with no hosts; the job later ends successful and notifies again.
+    make_cluster()
+
+    with mock.patch.object(Job, 'send_notification_templates') as send:
+        run_lost_instance_handling(mesh(PEER, EXEC))
+
+    assert mock.call('failed') not in send.call_args_list

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

@@ -62,10 +62,11 @@ class TestFindLostInstances:
             pytest.param(ME, 'control', 10_000, 0, False, id='this-instance-is-never-lost-here'),
             pytest.param(EXEC, 'execution', 150, 0, False, id='execution-node-gets-advertisement-period-added'),
             # A controller whose heartbeat is delayed past the grace period looks exactly like
-            # a dead one. The decision has only last_seen to go on.
+            # a dead one. The decision has only last_seen to go on (AAP-96824; the expected
+            # behavior is in test_scenarios_liveness.py).
             pytest.param(PEER, 'control', 125, 0, True, id='slow-controller-is-judged-lost-while-still-running'),
             # The peer heartbeated just now by its own clock, but this instance's clock runs
-            # five minutes ahead.
+            # five minutes ahead (AAP-96827; the expected behavior is in test_adoption_clock.py).
             pytest.param(PEER, 'control', 0, 300, True, id='clock-skew-on-the-judge-makes-a-healthy-peer-look-lost'),
         ],
     )
@@ -178,7 +179,8 @@ class TestAdoptionDeadline:
             pytest.param(None, 299, False, id='no-events-and-recent-start-is-within-deadline'),
             pytest.param(None, None, False, id='job-with-no-events-or-start-is-never-past-deadline'),
             # A playbook task that runs silently longer than the timeout (a long pause, a slow
-            # package install) looks the same as a dead one.
+            # package install) looks the same as a dead one (AAP-96829; the expected behavior
+            # is in test_scenarios_adoption.py).
             pytest.param(TIMEOUT + 1, TIMEOUT + 60, True, id='quiet-job-is-past-deadline-even-while-healthy'),
         ],
     )